[prabinovRedhat]

User description

Summary

This PR adds a new test TestConcurrentXcopyVddkMigration to validate the simultaneous execution of an XCOPY-based migration plan and a VDDK-based migration plan.

Key changes:

• Created TestConcurrentXcopyVddkMigration class in tests/test_copyoffload_migration.py.
• Added wait_for_concurrent_migration_execution utility in utilities/mtv_migration.py to handle simultaneous execution verification, replacing duplicated logic.
• Updated TestSimultaneousCopyoffloadMigrations to use the new utility.
• Added validation for offloadPlugin presence/absence in StorageMaps to distinguish between XCOPY and VDDK plans.
• Updated tests/tests_config/config.py with the new test configuration.

Test plan

• Run pytest tests/test_copyoffload_migration.py::TestConcurrentXcopyVddkMigration
• Verify that both plans execute simultaneously.
• Verify that the XCOPY plan has offloadPlugin configured in its StorageMap.
• Verify that the VDDK plan does NOT have offloadPlugin configured in its StorageMap.

---

PR Type

Enhancement, Tests

---

Description

• Add new test class TestConcurrentXcopyVddkMigration validating simultaneous XCOPY and VDDK migrations

• Extract concurrent migration validation logic into reusable wait_for_concurrent_migration_execution utility function

• Refactor TestSimultaneousCopyoffloadMigrations to use new shared utility, eliminating code duplication

• Add validation for offloadPlugin presence in XCOPY StorageMaps and absence in VDDK StorageMaps

---

Diagram Walkthrough

flowchart LR
  A["Test Class<br/>TestConcurrentXcopyVddkMigration"] -->|uses| B["wait_for_concurrent_migration_execution<br/>utility function"]
  C["TestSimultaneousCopyoffloadMigrations"] -->|refactored to use| B
  A -->|validates| D["XCOPY Plan<br/>with offloadPlugin"]
  A -->|validates| E["VDDK Plan<br/>without offloadPlugin"]
  B -->|monitors| F["Plan Status<br/>Executing/Succeeded/Failed"]

Loading

File Walkthrough

	Relevant files
Tests	test_copyoffload_migration.py Add concurrent XCOPY/VDDK test class and refactor validation logic tests/test_copyoffload_migration.py Added new test class TestConcurrentXcopyVddkMigration with 8 test methods for concurrent XCOPY and VDDK migration validation Removed inline concurrent migration validation logic (75 lines) from TestSimultaneousCopyoffloadMigrations.test_migrate_vms_simultaneously Replaced removed logic with call to new wait_for_concurrent_migration_execution utility function Added import for wait_for_concurrent_migration_execution from utilities Removed EARLY_COMPLETION_MSG error message template constant +312/-69
Configuration changes	config.py Add test configuration for concurrent XCOPY/VDDK migration tests/tests_config/config.py Added new test configuration test_concurrent_xcopy_vddk_migration with two VM definitions Configuration specifies warm_migration=False and copyoffload=True for test setup Defines separate VM templates for XCOPY and VDDK plan execution +20/-0
Enhancement	mtv_migration.py Add concurrent migration execution validation utility function utilities/mtv_migration.py Added new wait_for_concurrent_migration_execution function to handle simultaneous plan execution validation Function accepts list of Plan resources and optional timeout parameter (default 120 seconds) Implements polling logic to verify all plans reach "Executing" state before any completes Provides detailed error messages with status of all plans if early completion is detected Includes helper function _check_plan_status to determine plan status from conditions +62/-0

---

Summary by CodeRabbit

• Tests

  • Expanded test suite to validate concurrent XCOPY (copyoffload-enabled) and VDDK (non-offload) VM migrations end-to-end, including plan, network and storage setup, execution and post-migration checks.
  • Added orchestration to run migrations in parallel and assert they reach executing state simultaneously, with checks for disk counts and migration outcomes.
  • Added VM name handling and sanitized-name verification across XCOPY and VDDK flows.

• Chores

  • Added test configuration entries to support concurrent migration scenarios.

[redhat-qe-bot]

Report bugs in Issues

Welcome! 🎉

This pull request will be automatically processed with the following features:

🔄 Automatic Actions

• Reviewer Assignment: Reviewers are automatically assigned based on the OWNERS file in the repository root
• Size Labeling: PR size labels (XS, S, M, L, XL, XXL) are automatically applied based on changes
• Issue Creation: Disabled for this repository
• Branch Labeling: Branch-specific labels are applied to track the target branch
• Auto-verification: Auto-verified users have their PRs automatically marked as verified
• Labels: All label categories are enabled (default configuration)

📋 Available Commands

PR Status Management

• /wip - Mark PR as work in progress (adds WIP: prefix to title)
• /wip cancel - Remove work in progress status
• /hold - Block PR merging (approvers only)
• /hold cancel - Unblock PR merging
• /verified - Mark PR as verified
• /verified cancel - Remove verification status
• /reprocess - Trigger complete PR workflow reprocessing (useful if webhook failed or configuration changed)
• /regenerate-welcome - Regenerate this welcome message

Review & Approval

• /lgtm - Approve changes (looks good to me)
• /approve - Approve PR (approvers only)
• /automerge - Enable automatic merging when all requirements are met (maintainers and approvers only)
• /assign-reviewers - Assign reviewers based on OWNERS file
• /assign-reviewer @username - Assign specific reviewer
• /check-can-merge - Check if PR meets merge requirements

Testing & Validation

• /retest tox - Run Python test suite with tox
• /retest build-container - Rebuild and test container image
• /retest all - Run all available tests

Container Operations

• /build-and-push-container - Build and push container image (tagged with PR number)
  • Supports additional build arguments: /build-and-push-container --build-arg KEY=value

Cherry-pick Operations

• /cherry-pick <branch> - Schedule cherry-pick to target branch when PR is merged
  • Multiple branches: /cherry-pick branch1 branch2 branch3

Label Management

• /<label-name> - Add a label to the PR
• /<label-name> cancel - Remove a label from the PR

✅ Merge Requirements

This PR will be automatically approved when the following conditions are met:

1. Approval: /approve from at least one approver
2. LGTM Count: Minimum 0 /lgtm from reviewers
3. Status Checks: All required status checks must pass
4. No Blockers: No WIP, hold, conflict labels
5. Verified: PR must be marked as verified (if verification is enabled)

📊 Review Process

Approvers and Reviewers

Approvers:

• myakove

Reviewers:

• krcmarik
• myakove

Available Labels

• hold
• verified
• wip
• lgtm
• approve
• automerge

💡 Tips

• WIP Status: Use /wip when your PR is not ready for review
• Verification: The verified label is automatically removed on each new commit
• Cherry-picking: Cherry-pick labels are processed when the PR is merged
• Container Builds: Container images are automatically tagged with the PR number
• Permission Levels: Some commands require approver permissions
• Auto-verified Users: Certain users have automatic verification and merge privileges

For more information, please refer to the project documentation or contact the maintainers.

[qodo-code-review]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Robustness The new concurrent-execution polling assumes plan.instance.status.conditions exists and is iterable and that each condition has specific keys. If status/conditions are missing (common early in reconciliation or on transient API reads), this can raise AttributeError/TypeError/KeyError and fail tests for the wrong reason. Add defensive handling (e.g., treat missing conditions as "NotExecutingYet", guard with get, and/or catch exceptions) so the utility fails only on real early-completion or timeout. def wait_for_concurrent_migration_execution(plan_list: list[Plan], timeout: int = 120) -> None: """Wait for multiple migration plans to be executing simultaneously. Polls the status of all provided plans and validates that they all reach the "Executing" state at the same time. If any plan completes (Succeeded or Failed) before all plans are executing, the validation fails. Args: plan_list: List of Plan resources to monitor. timeout: Timeout in seconds to wait for simultaneous execution. Raises: AssertionError: If plans do not execute simultaneously or if any plan completes early. """ LOGGER.info(f"Validating {len(plan_list)} migrations enter executing state simultaneously") plans_executing = {plan.name: False for plan in plan_list} all_executing = False def _check_plan_status(plan: Plan) -> str: for cond in plan.instance.status.conditions: if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: cond_type = cond["type"] if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): return cond_type return "Executing" for sample in TimeoutSampler( func=lambda: {plan.name: _check_plan_status(plan) for plan in plan_list}, sleep=2, wait_timeout=timeout, ): current_statuses = sample # Update executing state for each plan for plan in plan_list: status = current_statuses[plan.name] if status == "Executing": if not plans_executing[plan.name]: LOGGER.info(f"Plan '{plan.name}' is now executing") plans_executing[plan.name] = True # Check for early completion failure elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): if not all_executing: # Construct error message with status of all plans status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) raise AssertionError( f"Plan {plan.name} reached {status} before all plans were executing simultaneously. " f"Statuses: {status_msg}" ) # Check if all plans are executing if all(plans_executing.values()): if not all_executing: LOGGER.info("SUCCESS: All migrations are executing simultaneously") all_executing = True break if not all_executing: raise AssertionError("Failed to validate all migrations executing simultaneously within timeout") Test Data The new test config defines two VMs with the same name value. If the framework expects two distinct VMs (e.g., one per plan) or uses names as identifiers, this can cause collisions, ambiguous inventory lookups, or both plans targeting the same VM set—making the concurrency assertion unreliable. Ensure the config clearly represents the intended XCOPY vs VDDK VM selections. "test_concurrent_xcopy_vddk_migration": { "virtual_machines": [ { "name": "xcopy-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, { "name": "xcopy-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, ], "warm_migration": False, "copyoffload": True, }, API Consistency The StorageMap offloadPlugin assertions mix attribute-style access (.offloadPlugin) and dict-style access (.get("offloadPlugin")) on spec.map[0]. Depending on the underlying object type returned by the dynamic client, one of these can break. Use a consistent access pattern and guard against empty spec.map to avoid IndexError masking real failures. assert self.storage_map_xcopy, "StorageMap creation failed for XCOPY plan" assert self.storage_map_xcopy.instance.spec.map[0].offloadPlugin, ( "XCOPY StorageMap missing offloadPlugin configuration" ) def test_create_storagemap_vddk( self, prepared_plan_2: dict[str, Any], fixture_store: dict[str, Any], ocp_admin_client: "DynamicClient", source_provider: BaseProvider, destination_provider: OCPProvider, source_provider_inventory: ForkliftInventory, target_namespace: str, ) -> None: """Create standard StorageMap for VDDK plan.""" vms_names = [vm["name"] for vm in prepared_plan_2["virtual_machines"]] self.__class__.storage_map_vddk = get_storage_migration_map( fixture_store=fixture_store, target_namespace=target_namespace, source_provider=source_provider, destination_provider=destination_provider, ocp_admin_client=ocp_admin_client, source_provider_inventory=source_provider_inventory, vms=vms_names, ) assert self.storage_map_vddk, "StorageMap creation failed for VDDK plan" assert not self.storage_map_vddk.instance.spec.map[0].get("offloadPlugin"), ( "VDDK StorageMap should NOT have offloadPlugin configuration" )

---

💡 Tool usage guide:

---

Overview:
The review tool scans the PR code changes, and generates a PR review which includes several types of feedbacks, such as possible PR issues, security threats and relevant test in the PR. More feedbacks can be added by configuring the tool.

The tool can be triggered automatically every time a new PR is opened, or can be invoked manually by commenting on any PR.

• When commenting, to edit configurations related to the review tool (pr_reviewer section), use the following template:

/review --pr_reviewer.some_config1=... --pr_reviewer.some_config2=...

• With a configuration file, use the following template:

[pr_reviewer]
some_config1=...
some_config2=...

See the review usage page for a comprehensive guide on using this tool.

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Walkthrough

Adds a new test class to run concurrent XCOPY (copyoffload) and VDDK migrations, a concurrency-wait utility to assert multiple Plans reach Executing simultaneously, and a test config entry for the concurrent migration scenario.

Changes

Cohort / File(s)	Summary
Test Orchestration tests/test_copyoffload_migration.py	Adds TestConcurrentXcopyVddkMigration with methods to create XCOPY/VDDK storagemaps and networkmaps, create plans, start concurrent migrations, replace inline concurrent checks with wait_for_concurrent_migration_execution, and verify migrated VMs (including disk counts and name sanitization).
Test Configuration tests/tests_config/config.py	Adds test_concurrent_xcopy_vddk_migration entry in tests_params defining two xcopy-template-test VMs (power: off, guest_agent: true, clone: true, disk_type: thick-lazy), with copyoffload: True and warm_migration: False.
Concurrency Validation Utility utilities/mtv_migration.py	Adds get_plan_migration_status(plan: Plan) -> str, refactors wait_for_migration_complate to use it, and introduces wait_for_concurrent_migration_execution(plan_list: list[Plan], timeout: int = 120) which polls multiple Plans and asserts they reach Executing concurrently (raises on early completion or timeout).

Sequence Diagram(s)

sequenceDiagram
  participant TestRunner as Test Runner
  participant API as OpenShift API
  participant SrcProv as Source Provider
  participant DstProv as Destination Provider

  TestRunner->>API: create storagemap (xcopy)
  TestRunner->>API: create storagemap (vddk)
  TestRunner->>API: create networkmap (xcopy)
  TestRunner->>API: create networkmap (vddk)
  TestRunner->>API: create plan_xcopy
  TestRunner->>API: create plan_vddk
  TestRunner->>API: start migration (plan_xcopy)
  TestRunner->>API: start migration (plan_vddk)
  loop poll (wait_for_concurrent_migration_execution)
    API->>TestRunner: return plan statuses
    TestRunner-->API: continue polling / assert states
  end
  TestRunner->>API: assert both plans = Executing
  API->>DstProv: provision migrated VMs
  TestRunner->>DstProv: validate migrated VMs (disk count, connectivity)

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

🚥 Pre-merge checks | ✅ 3

✅ Passed checks (3 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately and specifically summarizes the main change: adding a concurrent XCOPY and VDDK migration test, with the MTV-569 ticket reference providing clear context.
Docstring Coverage	✅ Passed	Docstring coverage is 87.50% which is sufficient. The required threshold is 80.00%.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[qodo-code-review]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Fragile Status Check The new concurrent-execution utility assumes the Plan status structure is always present and well-formed (e.g., status/conditions and required keys). In real clusters, these fields can be temporarily missing or partial; without defensive handling this can raise exceptions and create flaky tests. Consider guarding against missing status/conditions and malformed condition entries, and emitting a clear failure message when status cannot be determined. def wait_for_concurrent_migration_execution(plan_list: list[Plan], timeout: int = 120) -> None: """Wait for multiple migration plans to be executing simultaneously. Polls the status of all provided plans and validates that they all reach the "Executing" state at the same time. If any plan completes (Succeeded or Failed) before all plans are executing, the validation fails. Args: plan_list: List of Plan resources to monitor. timeout: Timeout in seconds to wait for simultaneous execution. Raises: AssertionError: If plans do not execute simultaneously or if any plan completes early. """ LOGGER.info(f"Validating {len(plan_list)} migrations enter executing state simultaneously") plans_executing = {plan.name: False for plan in plan_list} all_executing = False def _check_plan_status(plan: Plan) -> str: for cond in plan.instance.status.conditions: if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: cond_type = cond["type"] if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): return cond_type return "Executing" for sample in TimeoutSampler( func=lambda: {plan.name: _check_plan_status(plan) for plan in plan_list}, sleep=2, wait_timeout=timeout, ): current_statuses = sample # Update executing state for each plan for plan in plan_list: status = current_statuses[plan.name] if status == "Executing": if not plans_executing[plan.name]: LOGGER.info(f"Plan '{plan.name}' is now executing") plans_executing[plan.name] = True # Check for early completion failure elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): if not all_executing: # Construct error message with status of all plans status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) raise AssertionError( f"Plan {plan.name} reached {status} before all plans were executing simultaneously. " f"Statuses: {status_msg}" ) # Check if all plans are executing if all(plans_executing.values()): if not all_executing: LOGGER.info("SUCCESS: All migrations are executing simultaneously") all_executing = True break if not all_executing: raise AssertionError("Failed to validate all migrations executing simultaneously within timeout") OffloadPlugin Access The StorageMap assertions mix attribute access and dict-style access on spec.map[0]. Depending on the dynamic client object type, map[0] may not support .get() or direct attribute access, which can cause runtime errors. Use a consistent, safe access pattern (e.g., getattr(..., "offloadPlugin", None) or normalize to dict) before asserting presence/absence. self.__class__.storage_map_xcopy = get_storage_migration_map( fixture_store=fixture_store, target_namespace=target_namespace, source_provider=source_provider, destination_provider=destination_provider, ocp_admin_client=ocp_admin_client, source_provider_inventory=source_provider_inventory, vms=vms_names, storage_class=storage_class, datastore_id=datastore_id, offload_plugin_config=offload_plugin_config, access_mode="ReadWriteOnce", volume_mode="Block", ) assert self.storage_map_xcopy, "StorageMap creation failed for XCOPY plan" assert self.storage_map_xcopy.instance.spec.map[0].offloadPlugin, ( "XCOPY StorageMap missing offloadPlugin configuration" ) def test_create_storagemap_vddk( self, prepared_plan_2: dict[str, Any], fixture_store: dict[str, Any], ocp_admin_client: "DynamicClient", source_provider: BaseProvider, destination_provider: OCPProvider, source_provider_inventory: ForkliftInventory, target_namespace: str, ) -> None: """Create standard StorageMap for VDDK plan.""" vms_names = [vm["name"] for vm in prepared_plan_2["virtual_machines"]] self.__class__.storage_map_vddk = get_storage_migration_map( fixture_store=fixture_store, target_namespace=target_namespace, source_provider=source_provider, destination_provider=destination_provider, ocp_admin_client=ocp_admin_client, source_provider_inventory=source_provider_inventory, vms=vms_names, ) assert self.storage_map_vddk, "StorageMap creation failed for VDDK plan" assert not self.storage_map_vddk.instance.spec.map[0].get("offloadPlugin"), ( "VDDK StorageMap should NOT have offloadPlugin configuration" ) Test Data Collision The new test config defines two VMs with the same name. If the test framework does not automatically disambiguate/clones names, this can lead to selection ambiguity, inventory lookup issues, or resource name collisions. Ensure the two entries are intentionally distinct (different template names or explicit unique identifiers) or document/encode the disambiguation behavior in the fixture. "test_concurrent_xcopy_vddk_migration": { "virtual_machines": [ { "name": "xcopy-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, { "name": "xcopy-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, ], "warm_migration": False, "copyoffload": True, },

---

💡 Tool usage guide:

---

Overview:
The review tool scans the PR code changes, and generates a PR review which includes several types of feedbacks, such as possible PR issues, security threats and relevant test in the PR. More feedbacks can be added by configuring the tool.

The tool can be triggered automatically every time a new PR is opened, or can be invoked manually by commenting on any PR.

• When commenting, to edit configurations related to the review tool (pr_reviewer section), use the following template:

/review --pr_reviewer.some_config1=... --pr_reviewer.some_config2=...

• With a configuration file, use the following template:

[pr_reviewer]
some_config1=...
some_config2=...

See the review usage page for a comprehensive guide on using this tool.

[qodo-code-review]

PR Code Suggestions ✨

Latest suggestions up to 99d345d

Category	Suggestion	Impact
Possible bug	Ensure unique VM identifiers In the test_concurrent_xcopy_vddk_migration configuration, assign unique names to the two virtual machines to prevent identifier collisions and ensure tests deterministically target different VMs. tests/tests_config/config.py [473-492] "test_concurrent_xcopy_vddk_migration": { "virtual_machines": [ { - "name": "xcopy-template-test", + "name": "xcopy-template-test-1", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, { - "name": "xcopy-template-test", + "name": "xcopy-template-test-2", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, ], "warm_migration": False, "copyoffload": True, }, Apply / Chat Suggestion importance[1-10]: 9 __ Why: This suggestion correctly identifies a critical bug where two VMs in the test configuration share the same name. Since the name is used as a unique identifier for lookups (e.g., in ForkliftInventory.get_vm), this would cause tests to be flaky or incorrect by targeting the same VM twice instead of two distinct ones.	High
	Guard against empty mappings Before accessing the first element of self.storage_map_xcopy.instance.spec.map, add a check to ensure the list is not empty to prevent potential IndexError exceptions and improve error reporting. tests/test_copyoffload_migration.py [3269-3272] assert self.storage_map_xcopy, "StorageMap creation failed for XCOPY plan" -assert self.storage_map_xcopy.instance.spec.map[0].get("offloadPlugin"), ( - "XCOPY StorageMap missing offloadPlugin configuration" -) +storage_map_entries = (self.storage_map_xcopy.instance.spec.map or []) if self.storage_map_xcopy.instance.spec else [] +assert storage_map_entries, "XCOPY StorageMap has empty mapping list" +assert storage_map_entries[0].get("offloadPlugin"), "XCOPY StorageMap missing offloadPlugin configuration" Apply / Chat Suggestion importance[1-10]: 6 __ Why: The suggestion correctly points out a potential IndexError if self.storage_map_xcopy.instance.spec.map is empty. Adding a check for an empty map before indexing improves the robustness of the test and provides a clearer error message if the StorageMap is created without any mappings.	Low
Maintainability	Refresh and harden status parsing In get_plan_migration_status, refresh the plan object before accessing its status to prevent stale data during polling. Additionally, use .get() when accessing keys in the conditions dictionary to avoid KeyError exceptions. utilities/mtv_migration.py [308-338] def get_plan_migration_status(plan: Plan) -> str: """Get the migration status from the Plan conditions. Args: plan (Plan): The Plan resource to check. Returns: str: The status of the plan ("Executing", "Succeeded", "Failed", or empty string if unknown). Raises: None """ + plan.refresh() status = plan.instance.status if not status: return "" conditions = status.conditions - if conditions: - for cond in conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] - - if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - return cond_type + for cond in conditions or []: + if cond.get("category") == "Advisory" and cond.get("status") == Plan.Condition.Status.TRUE: + cond_type = cond.get("type") + if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): + return cond_type # Check if Migration CR exists to confirm execution try: _find_migration_for_plan(plan) return Plan.Status.EXECUTING except MigrationNotFoundError: return "" Apply / Chat Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that polling the plan.instance without refreshing can lead to stale data. Adding plan.refresh() and using .get() for dictionary access makes the status checking more robust and resilient against race conditions and unexpected API responses, which is important in a concurrent testing context.	Medium
More

---

Previous suggestions

Suggestions up to commit cb77c50

Category	Suggestion	Impact
Possible bug	Ensure distinct plans are validated In wait_for_concurrent_migration_execution, ensure the input plan_list contains at least two distinct plans to prevent false positives when validating concurrent execution. utilities/mtv_migration.py [570-617] def wait_for_concurrent_migration_execution(plan_list: list[Plan], timeout: int = 120) -> None: ... - plans_executing = {plan.name: False for plan in plan_list} + unique_plans: list[Plan] = [] + seen: set[tuple[str, str]] = set() + for plan in plan_list: + key = (plan.namespace, plan.name) + if key not in seen: + seen.add(key) + unique_plans.append(plan) + + if len(unique_plans) < 2: + raise AssertionError("Concurrent migration validation requires at least two distinct plans") + + plans_executing = {(plan.namespace, plan.name): False for plan in unique_plans} try: for current_statuses in TimeoutSampler( - func=lambda: {plan.name: get_plan_migration_status(plan) for plan in plan_list}, + func=lambda: {(p.namespace, p.name): get_plan_migration_status(p) for p in unique_plans}, sleep=2, wait_timeout=timeout, ): - # Update executing state for each plan - for plan in plan_list: - status = current_statuses[plan.name] + for plan in unique_plans: + key = (plan.namespace, plan.name) + status = current_statuses[key] if status == Plan.Status.EXECUTING: - if not plans_executing[plan.name]: + if not plans_executing[key]: LOGGER.info(f"Plan '{plan.name}' is now executing") - plans_executing[plan.name] = True + plans_executing[key] = True continue - # Check for early completion failure - elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - # Construct error message with status of all plans - status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) + if status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): + status_msg = ", ".join(f"{ns}/{name}: {stat}" for (ns, name), stat in current_statuses.items()) raise AssertionError( f"Plan {plan.name} reached {status} before all plans were executing simultaneously. " f"Statuses: {status_msg}" ) - # Check if all plans are executing if all(plans_executing.values()): LOGGER.info("SUCCESS: All migrations are executing simultaneously") return Suggestion importance[1-10]: 8 __ Why: This suggestion correctly identifies a logical flaw where the function would incorrectly report success if given duplicate plans, and it provides a robust fix by ensuring only unique plans are validated.	Medium
Performance	Refresh status and cache lookups In get_plan_migration_status, call plan.refresh() to get the latest status and cache the result of _find_migration_for_plan to improve performance and reduce API load. utilities/mtv_migration.py [308-338] +_PLAN_HAS_MIGRATION: dict[tuple[str, str], bool] = {} + + def get_plan_migration_status(plan: Plan) -> str: ... + plan.refresh() + status = plan.instance.status if not status: return "" conditions = getattr(status, "conditions", None) if conditions: for cond in conditions: if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: cond_type = cond["type"] - if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): return cond_type - # Check if Migration CR exists to confirm execution + # Infer executing state with cached Migration existence to reduce API calls + cache_key = (plan.namespace, plan.name) + if _PLAN_HAS_MIGRATION.get(cache_key): + return Plan.Status.EXECUTING + try: _find_migration_for_plan(plan) + _PLAN_HAS_MIGRATION[cache_key] = True return Plan.Status.EXECUTING except MigrationNotFoundError: return "" Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that polling plan.instance.status without a refresh can lead to reading stale data, and calling plan.refresh() fixes this potential correctness issue. Caching the migration lookup is a good performance optimization.	Medium
Maintainability	Use distinct VM template names In the test_concurrent_xcopy_vddk_migration configuration, use distinct names for the two virtual machines to avoid ambiguity and potential collisions during testing. tests/tests_config/config.py [473-492] "test_concurrent_xcopy_vddk_migration": { "virtual_machines": [ { "name": "xcopy-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, { - "name": "xcopy-template-test", + "name": "vddk-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, ], "warm_migration": False, "copyoffload": True, }, Suggestion importance[1-10]: 6 __ Why: The suggestion correctly points out that using identical source VM names for parallel tests can cause ambiguity, and using distinct names improves test clarity and maintainability.	Low
best practice	Fail fast on missing config In test_create_storagemap_xcopy, use py_config.get("storage_class") and explicitly check for None to provide a clearer error message if the configuration key is missing. tests/test_copyoffload_migration.py [3244] -storage_class = py_config["storage_class"] +storage_class = py_config.get("storage_class") +if storage_class is None: + raise ValueError("Missing required config key: 'storage_class'") Suggestion importance[1-10]: 5 __ Why: The suggestion improves robustness by replacing direct dictionary access with .get() and adding an explicit check, which provides clearer error messages for missing configuration.	Low

✅ Suggestions up to commit 627d1fe

Category	Suggestion	Impact
Maintainability	✅ Harden status parsing logic Suggestion Impact: The commit modified get_plan_migration_status to be more defensive by guarding the conditions iteration (only looping when conditions is truthy) and changing how unknown/pending states are represented, plus adding an execution confirmation via checking for a Migration CR; however, it did not implement the suggested .get()-based key access or dict-type handling for conditions. code diff: @@ -312,27 +312,30 @@ plan (Plan): The Plan resource to check. Returns: - str: The status of the plan ("Pending", "Executing", "Succeeded", or "Failed"). + str: The status of the plan ("Executing", "Succeeded", "Failed", or empty string if unknown). Raises: None """ - status = getattr(plan.instance, "status", None) + status = plan.instance.status if not status: - return "Pending" + return "" conditions = getattr(status, "conditions", None) - if not conditions: - return "Pending" - - for cond in conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] - - if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - return cond_type - - return "Executing" + if conditions: + for cond in conditions: + if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: + cond_type = cond["type"] + + if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): + return cond_type + + # Check if Migration CR exists to confirm execution + try: + _find_migration_for_plan(plan) + return Plan.Status.EXECUTING + except MigrationNotFoundError: + return "" Refactor get_plan_migration_status to safely handle dictionary access using .get() and check for unexpected data types for the conditions field. This will make the function more resilient to malformed or incomplete API responses. utilities/mtv_migration.py [308-335] def get_plan_migration_status(plan: Plan) -> str: """Get the migration status from the Plan conditions. Args: plan (Plan): The Plan resource to check. Returns: str: The status of the plan ("Pending", "Executing", "Succeeded", or "Failed"). - - Raises: - None """ status = getattr(plan.instance, "status", None) - if not status: + if status is None: return "Pending" - conditions = getattr(status, "conditions", None) - if not conditions: - return "Pending" + conditions = getattr(status, "conditions", None) or [] + if isinstance(conditions, dict): + conditions = conditions.values() for cond in conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] - - if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - return cond_type + if ( + cond.get("category") == "Advisory" + and cond.get("status") == Plan.Condition.Status.TRUE + and cond.get("type") in (Plan.Status.SUCCEEDED, Plan.Status.FAILED) + ): + return cond["type"] return "Executing" Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that the get_plan_migration_status function can crash due to missing keys or unexpected data types from the API. The proposed changes make the status polling more resilient, preventing test failures caused by transient or malformed API responses.	Medium
Possible bug	Ensure unique VM identities In the test_concurrent_xcopy_vddk_migration configuration, assign unique names to the two virtual_machines entries to prevent potential ambiguity and resource collisions during testing. tests/tests_config/config.py [473-492] "test_concurrent_xcopy_vddk_migration": { "virtual_machines": [ { - "name": "xcopy-template-test", + "name": "xcopy-template-test-xcopy", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, { - "name": "xcopy-template-test", + "name": "xcopy-template-test-vddk", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, ], "warm_migration": False, "copyoffload": True, }, Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that using identical name values for VMs in the test configuration is poor practice and can lead to ambiguity, even if the test framework's cloning mechanism currently prevents a bug. Adopting unique names makes the test configuration clearer and more robust against future changes.	Low
	Avoid unsafe list indexing Add a check to ensure self.storage_map_xcopy.instance.spec.map is not empty before accessing its first element to avoid a potential IndexError. tests/test_copyoffload_migration.py [3270-3272] -assert self.storage_map_xcopy.instance.spec.map[0].get("offloadPlugin"), ( - "XCOPY StorageMap missing offloadPlugin configuration" -) +storage_map_entries = self.storage_map_xcopy.instance.spec.map or [] +assert storage_map_entries, "XCOPY StorageMap has no mapping entries" +assert storage_map_entries[0].get("offloadPlugin"), "XCOPY StorageMap missing offloadPlugin configuration" Suggestion importance[1-10]: 5 __ Why: This suggestion improves test robustness by preventing an IndexError if the StorageMap is created without any mappings. It replaces a potential crash with a clear assertion failure, which is better for test reliability and debugging.	Low

✅ Suggestions up to commit db02ea8

Category	Suggestion	Impact
Possible bug	✅ Key concurrent tracking by namespace Suggestion Impact: The commit refactored concurrent plan status polling to use a shared get_plan_migration_status helper and adjusted the loop to return on success and raise on timeout, aligning with parts of the suggestion. However, it did not implement (namespace, name) tuple keys nor add the "at least two plans" validation. code diff: +def get_plan_migration_status(plan: Plan) -> str: + """Get the migration status from the Plan conditions. + + Args: + plan (Plan): The Plan resource to check. + + Returns: + str: The status of the plan ("Pending", "Executing", "Succeeded", or "Failed"). + """ + status = getattr(plan.instance, "status", None) + if not status: + return "Pending" + + conditions = getattr(status, "conditions", None) + if not conditions: + return "Pending" + + for cond in conditions: + if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: + cond_type = cond["type"] + + if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): + return cond_type + + return "Executing" + + def wait_for_migration_complate(plan: Plan) -> None: - def _wait_for_migration_complate(_plan: Plan) -> str: - for cond in _plan.instance.status.conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] - - if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - return cond_type - - return "Executing" - try: last_status: str = "" for sample in TimeoutSampler( - func=_wait_for_migration_complate, + func=get_plan_migration_status, sleep=1, wait_timeout=py_config.get("plan_wait_timeout", 600), - _plan=plan, + plan=plan, ): if sample != last_status: LOGGER.info(f"Plan '{plan.name}' migration status: '{sample}'") @@ -555,39 +572,32 @@ plan_list: List of Plan resources to monitor. timeout: Timeout in seconds to wait for simultaneous execution. + Returns: + None + Raises: AssertionError: If plans do not execute simultaneously or if any plan completes early. """ LOGGER.info(f"Validating {len(plan_list)} migrations enter executing state simultaneously") plans_executing = {plan.name: False for plan in plan_list} - all_executing = False - - def _check_plan_status(plan: Plan) -> str: - for cond in plan.instance.status.conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] - if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - return cond_type - return "Executing" - - for sample in TimeoutSampler( - func=lambda: {plan.name: _check_plan_status(plan) for plan in plan_list}, - sleep=2, - wait_timeout=timeout, - ): - current_statuses = sample - - # Update executing state for each plan - for plan in plan_list: - status = current_statuses[plan.name] - if status == "Executing": - if not plans_executing[plan.name]: - LOGGER.info(f"Plan '{plan.name}' is now executing") - plans_executing[plan.name] = True - - # Check for early completion failure - elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - if not all_executing: + + try: + for current_statuses in TimeoutSampler( + func=lambda: {plan.name: get_plan_migration_status(plan) for plan in plan_list}, + sleep=2, + wait_timeout=timeout, + ): + # Update executing state for each plan + for plan in plan_list: + status = current_statuses[plan.name] + if status == "Executing": + if not plans_executing[plan.name]: + LOGGER.info(f"Plan '{plan.name}' is now executing") + plans_executing[plan.name] = True + continue + + # Check for early completion failure + elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): # Construct error message with status of all plans status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) raise AssertionError( @@ -595,13 +605,11 @@ f"Statuses: {status_msg}" ) - # Check if all plans are executing - if all(plans_executing.values()): - if not all_executing: + # Check if all plans are executing + if all(plans_executing.values()): LOGGER.info("SUCCESS: All migrations are executing simultaneously") - all_executing = True - break - - if not all_executing: - raise AssertionError("Failed to validate all migrations executing simultaneously within timeout") - + return + + except TimeoutExpiredError: + raise AssertionError("Failed to validate all migrations executing simultaneously within timeout") from None In wait_for_concurrent_migration_execution, use a (namespace, name) tuple as the unique key for tracking plans instead of just plan.name to prevent collisions. Also, add a validation check to ensure at least two plans are provided. utilities/mtv_migration.py [564-614] def wait_for_concurrent_migration_execution(plan_list: list[Plan], timeout: int = 120) -> None: """Wait for multiple migration plans to be executing simultaneously. Polls the status of all provided plans and validates that they all reach the "Executing" state at the same time. If any plan completes (Succeeded or Failed) before all plans are executing, the validation fails. - ... + + Args: + plan_list: List of Plan resources to monitor. + timeout: Timeout in seconds to wait for simultaneous execution. + + Returns: + None + + Raises: + AssertionError: If plans do not execute simultaneously or if any plan completes early. """ + if len(plan_list) < 2: + raise AssertionError("Concurrent migration validation requires at least two plans") + LOGGER.info(f"Validating {len(plan_list)} migrations enter executing state simultaneously") - plans_executing = {plan.name: False for plan in plan_list} + plan_keys = [(p.namespace, p.name) for p in plan_list] + plans_executing = {key: False for key in plan_keys} + + def _statuses() -> dict[tuple[str, str], str]: + return {(p.namespace, p.name): get_plan_migration_status(p) for p in plan_list} try: for current_statuses in TimeoutSampler( - func=lambda: {plan.name: get_plan_migration_status(plan) for plan in plan_list}, + func=_statuses, sleep=2, wait_timeout=timeout, ): - # Update executing state for each plan for plan in plan_list: - status = current_statuses[plan.name] + key = (plan.namespace, plan.name) + status = current_statuses[key] + if status == "Executing": - if not plans_executing[plan.name]: + if not plans_executing[key]: LOGGER.info(f"Plan '{plan.name}' is now executing") - plans_executing[plan.name] = True + plans_executing[key] = True continue - # Check for early completion failure - elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - # Construct error message with status of all plans - status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) + if status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): + status_msg = ", ".join([f"{ns}/{name}: {stat}" for (ns, name), stat in current_statuses.items()]) raise AssertionError( - f"Plan {plan.name} reached {status} before all plans were executing simultaneously. " + f"Plan {plan.namespace}/{plan.name} reached {status} before all plans were executing simultaneously. " f"Statuses: {status_msg}" ) - # Check if all plans are executing if all(plans_executing.values()): LOGGER.info("SUCCESS: All migrations are executing simultaneously") return except TimeoutExpiredError: raise AssertionError("Failed to validate all migrations executing simultaneously within timeout") from None Suggestion importance[1-10]: 7 __ Why: This suggestion correctly points out that resource names are only unique within a namespace. Using a (namespace, name) tuple as a key makes the utility function more robust and correct, preventing potential bugs if it's used with plans from different namespaces that share the same name.	Medium
	✅ Refresh and harden status parsing Suggestion Impact: The PR refactored status parsing into a shared get_plan_migration_status() helper and updated polling callers to use it, aligning with the suggestion’s goal of centralizing/hardening status parsing. However, the commit did not add the proposed plan.refresh() call and still uses cond["..."] indexing rather than cond.get(...). code diff: +def get_plan_migration_status(plan: Plan) -> str: + """Get the migration status from the Plan conditions. + + Args: + plan (Plan): The Plan resource to check. + + Returns: + str: The status of the plan ("Pending", "Executing", "Succeeded", or "Failed"). + """ + status = getattr(plan.instance, "status", None) + if not status: + return "Pending" + + conditions = getattr(status, "conditions", None) + if not conditions: + return "Pending" + + for cond in conditions: + if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: + cond_type = cond["type"] + + if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): + return cond_type + + return "Executing" + + def wait_for_migration_complate(plan: Plan) -> None: - def _wait_for_migration_complate(_plan: Plan) -> str: - for cond in _plan.instance.status.conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] - - if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - return cond_type - - return "Executing" - try: last_status: str = "" for sample in TimeoutSampler( - func=_wait_for_migration_complate, + func=get_plan_migration_status, sleep=1, wait_timeout=py_config.get("plan_wait_timeout", 600), - _plan=plan, + plan=plan, ): if sample != last_status: LOGGER.info(f"Plan '{plan.name}' migration status: '{sample}'") @@ -555,39 +572,32 @@ plan_list: List of Plan resources to monitor. timeout: Timeout in seconds to wait for simultaneous execution. + Returns: + None + Raises: AssertionError: If plans do not execute simultaneously or if any plan completes early. """ LOGGER.info(f"Validating {len(plan_list)} migrations enter executing state simultaneously") plans_executing = {plan.name: False for plan in plan_list} - all_executing = False - - def _check_plan_status(plan: Plan) -> str: - for cond in plan.instance.status.conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] - if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - return cond_type - return "Executing" - - for sample in TimeoutSampler( - func=lambda: {plan.name: _check_plan_status(plan) for plan in plan_list}, - sleep=2, - wait_timeout=timeout, - ): - current_statuses = sample - - # Update executing state for each plan - for plan in plan_list: - status = current_statuses[plan.name] - if status == "Executing": - if not plans_executing[plan.name]: - LOGGER.info(f"Plan '{plan.name}' is now executing") - plans_executing[plan.name] = True - - # Check for early completion failure - elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - if not all_executing: + + try: + for current_statuses in TimeoutSampler( + func=lambda: {plan.name: get_plan_migration_status(plan) for plan in plan_list}, + sleep=2, + wait_timeout=timeout, + ): + # Update executing state for each plan In get_plan_migration_status, refresh the plan object before checking its status and use .get() for safer access to condition properties. This prevents flaky tests from stale data and avoids KeyError exceptions. utilities/mtv_migration.py [308-332] def get_plan_migration_status(plan: Plan) -> str: """Get the migration status from the Plan conditions. Args: plan (Plan): The Plan resource to check. Returns: str: The status of the plan ("Pending", "Executing", "Succeeded", or "Failed"). """ + # Best-effort refresh to avoid sampling a stale cached instance during polling loops. + refresh = getattr(plan, "refresh", None) + if callable(refresh): + refresh() + status = getattr(plan.instance, "status", None) - if not status: + if status is None: return "Pending" conditions = getattr(status, "conditions", None) if not conditions: return "Pending" for cond in conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] + if cond.get("category") != "Advisory": + continue + if cond.get("status") != Plan.Condition.Status.TRUE: + continue - if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - return cond_type + cond_type = cond.get("type") + if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): + return cond_type return "Executing" Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that polling without refreshing the resource can lead to using stale data, making tests flaky. Using .get() also improves robustness against unexpected changes in the API response, preventing potential KeyError exceptions.	Low

✅ Suggestions up to commit 92a786d

Category	Suggestion	Impact
Possible bug	Avoid VM name collisions in config In the test_concurrent_xcopy_vddk_migration configuration, use distinct names for the two virtual machines to prevent potential naming collisions and ensure the concurrency test operates on separate, unique VMs. tests/tests_config/config.py [473-492] "test_concurrent_xcopy_vddk_migration": { "virtual_machines": [ { - "name": "xcopy-template-test", + "name": "xcopy-template-test-1", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, { - "name": "xcopy-template-test", + "name": "xcopy-template-test-2", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, ], "warm_migration": False, "copyoffload": True, }, Suggestion importance[1-10]: 8 __ Why: This is a valid and important catch. Using the same source VM name for two plans, even with cloning, can lead to race conditions and flaky tests if the test framework's cloning and naming logic isn't perfectly deterministic and isolated. Using distinct source names makes the test more robust and easier to debug.	Medium
	✅ Handle missing status conditions safely Suggestion Impact: Added defensive checks for missing plan.instance.status and status.conditions, returning "Executing" when absent instead of iterating unsafely. code diff: + status = getattr(plan.instance, "status", None) + if not status: + return "Executing" + + conditions = getattr(status, "conditions", None) + if not conditions: + return "Executing" + + for cond in conditions: Refactor get_plan_migration_status to safely handle cases where plan.instance.status.conditions is missing. Use .get() with default values to prevent potential AttributeError or TypeError and improve the function's robustness. utilities/mtv_migration.py [308-324] def get_plan_migration_status(plan: Plan) -> str: """Get the migration status from the Plan conditions. Args: plan (Plan): The Plan resource to check. Returns: str: The status of the plan ("Executing", "Succeeded", or "Failed"). """ - for cond in plan.instance.status.conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] + conditions = (plan.instance.get("status", {}) or {}).get("conditions") or [] + for cond in conditions: + if cond.get("category") == "Advisory" and cond.get("status") == Plan.Condition.Status.TRUE: + cond_type = cond.get("type") if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): return cond_type return "Executing" Suggestion importance[1-10]: 8 __ Why: This suggestion correctly identifies a potential AttributeError or TypeError if plan.instance.status.conditions is not present, which is a common scenario in Kubernetes object lifecycles. The proposed change makes the polling logic more resilient to transient API states, preventing test failures for non-product reasons, which is a critical improvement for test stability.	Medium
Possible issue	Use unique keys for plans In wait_for_concurrent_migration_execution, use a composite key of "{namespace}/{name}" instead of just plan.name for dictionaries. This ensures uniqueness when handling plans from different namespaces, preventing potential status overwrites. utilities/mtv_migration.py [556-606] def wait_for_concurrent_migration_execution(plan_list: list[Plan], timeout: int = 120) -> None: """Wait for multiple migration plans to be executing simultaneously. Polls the status of all provided plans and validates that they all reach the "Executing" state at the same time. If any plan completes (Succeeded or Failed) before all plans are executing, the validation fails. Args: plan_list: List of Plan resources to monitor. timeout: Timeout in seconds to wait for simultaneous execution. Returns: None Raises: AssertionError: If plans do not execute simultaneously or if any plan completes early. """ + def _plan_key(plan: Plan) -> str: + return f"{plan.namespace}/{plan.name}" + LOGGER.info(f"Validating {len(plan_list)} migrations enter executing state simultaneously") - plans_executing = {plan.name: False for plan in plan_list} + plans_executing = {_plan_key(plan): False for plan in plan_list} try: for current_statuses in TimeoutSampler( - func=lambda: {plan.name: get_plan_migration_status(plan) for plan in plan_list}, + func=lambda: {_plan_key(plan): get_plan_migration_status(plan) for plan in plan_list}, sleep=2, wait_timeout=timeout, ): - # Update executing state for each plan for plan in plan_list: - status = current_statuses[plan.name] + key = _plan_key(plan) + status = current_statuses[key] if status == "Executing": - if not plans_executing[plan.name]: - LOGGER.info(f"Plan '{plan.name}' is now executing") - plans_executing[plan.name] = True + if not plans_executing[key]: + LOGGER.info(f"Plan '{key}' is now executing") + plans_executing[key] = True continue - # Check for early completion failure - elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - # Construct error message with status of all plans + if status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) raise AssertionError( - f"Plan {plan.name} reached {status} before all plans were executing simultaneously. " + f"Plan {key} reached {status} before all plans were executing simultaneously. " f"Statuses: {status_msg}" ) - # Check if all plans are executing if all(plans_executing.values()): LOGGER.info("SUCCESS: All migrations are executing simultaneously") return except TimeoutExpiredError: raise AssertionError("Failed to validate all migrations executing simultaneously within timeout") from None Suggestion importance[1-10]: 6 __ Why: The suggestion correctly points out that using only plan.name as a key is not unique across namespaces. While the current usage in the PR seems safe as plans are in the same namespace, using a namespace/name key makes the utility function more robust and reusable for future scenarios involving multiple namespaces.	Low
Enhancement	Ensure copy-offload config is absent In test_create_storagemap_vddk, change the assertion to check for the absence of the offloadPlugin key using not in. This provides a stricter check than verifying a falsy value, ensuring no offload configuration exists. tests/test_copyoffload_migration.py [3310-3312] -assert not self.storage_map_vddk.instance.spec.map[0].get("offloadPlugin"), ( +assert "offloadPlugin" not in self.storage_map_vddk.instance.spec.map[0], ( "VDDK StorageMap should NOT have offloadPlugin configuration" ) Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that checking for the absence of a key is more precise than checking for a falsy value. This improves the assertion's strictness, ensuring that no offloadPlugin configuration, even an empty one, is present for VDDK plans.	Low

✅ Suggestions up to commit 98be08a

Category	Suggestion	Impact
Possible bug	✅ Harden concurrent status polling Suggestion Impact: The PR changes refactored concurrent polling to use a shared helper (get_plan_migration_status) and adjusted timeout handling (catching TimeoutExpiredError and returning on success), but it did not implement the core hardening requested: no plan.get() refresh per poll and no guard for absent status/conditions. code diff: +def get_plan_migration_status(plan: Plan) -> str: + """Get the migration status from the Plan conditions. + + Args: + plan (Plan): The Plan resource to check. + + Returns: + str: The status of the plan ("Executing", "Succeeded", or "Failed"). + """ + for cond in plan.instance.status.conditions: + if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: + cond_type = cond["type"] + + if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): + return cond_type + + return "Executing" + + def wait_for_migration_complate(plan: Plan) -> None: - def _wait_for_migration_complate(_plan: Plan) -> str: - for cond in _plan.instance.status.conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] - - if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - return cond_type - - return "Executing" - try: last_status: str = "" for sample in TimeoutSampler( - func=_wait_for_migration_complate, + func=get_plan_migration_status, sleep=1, wait_timeout=py_config.get("plan_wait_timeout", 600), - _plan=plan, + plan=plan, ): if sample != last_status: LOGGER.info(f"Plan '{plan.name}' migration status: '{sample}'") @@ -563,33 +572,24 @@ """ LOGGER.info(f"Validating {len(plan_list)} migrations enter executing state simultaneously") plans_executing = {plan.name: False for plan in plan_list} - all_executing = False - - def _check_plan_status(plan: Plan) -> str: - for cond in plan.instance.status.conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] - if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - return cond_type - return "Executing" - - for current_statuses in TimeoutSampler( - func=lambda: {plan.name: _check_plan_status(plan) for plan in plan_list}, - sleep=2, - wait_timeout=timeout, - ): - # Update executing state for each plan - for plan in plan_list: - status = current_statuses[plan.name] - if status == "Executing": - if not plans_executing[plan.name]: - LOGGER.info(f"Plan '{plan.name}' is now executing") - plans_executing[plan.name] = True - continue - - # Check for early completion failure - elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - if not all_executing: + + try: + for current_statuses in TimeoutSampler( + func=lambda: {plan.name: get_plan_migration_status(plan) for plan in plan_list}, + sleep=2, + wait_timeout=timeout, + ): + # Update executing state for each plan + for plan in plan_list: + status = current_statuses[plan.name] + if status == "Executing": + if not plans_executing[plan.name]: + LOGGER.info(f"Plan '{plan.name}' is now executing") + plans_executing[plan.name] = True + continue + + # Check for early completion failure + elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): # Construct error message with status of all plans status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) raise AssertionError( @@ -597,13 +597,11 @@ f"Statuses: {status_msg}" ) - # Check if all plans are executing - if all(plans_executing.values()): - if not all_executing: + # Check if all plans are executing + if all(plans_executing.values()): LOGGER.info("SUCCESS: All migrations are executing simultaneously") - all_executing = True - break - - if not all_executing: - raise AssertionError("Failed to validate all migrations executing simultaneously within timeout") - + return + + except TimeoutExpiredError: + raise AssertionError("Failed to validate all migrations executing simultaneously within timeout") from None + In wait_for_concurrent_migration_execution, refresh the Plan object on each poll to get the latest status. Also, handle cases where status.conditions is not yet available to prevent errors and incorrect status reporting. utilities/mtv_migration.py [547-608] def wait_for_concurrent_migration_execution(plan_list: list[Plan], timeout: int = 120) -> None: - """Wait for multiple migration plans to be executing simultaneously. - - Polls the status of all provided plans and validates that they all reach the "Executing" - state at the same time. If any plan completes (Succeeded or Failed) before all plans - are executing, the validation fails. - - Args: - plan_list: List of Plan resources to monitor. - timeout: Timeout in seconds to wait for simultaneous execution. - - Returns: - None - - Raises: - AssertionError: If plans do not execute simultaneously or if any plan completes early. - """ LOGGER.info(f"Validating {len(plan_list)} migrations enter executing state simultaneously") plans_executing = {plan.name: False for plan in plan_list} all_executing = False def _check_plan_status(plan: Plan) -> str: - for cond in plan.instance.status.conditions: - if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: - cond_type = cond["type"] + plan.get() + conditions: list[dict[str, Any]] = (plan.instance.status or {}).get("conditions") or [] + for cond in conditions: + if cond.get("category") == "Advisory" and cond.get("status") == Plan.Condition.Status.TRUE: + cond_type = cond.get("type") if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): return cond_type - return "Executing" + + # Not completed and no completion condition observed yet + return "Pending" for current_statuses in TimeoutSampler( func=lambda: {plan.name: _check_plan_status(plan) for plan in plan_list}, sleep=2, wait_timeout=timeout, ): - # Update executing state for each plan for plan in plan_list: status = current_statuses[plan.name] - if status == "Executing": - if not plans_executing[plan.name]: - LOGGER.info(f"Plan '{plan.name}' is now executing") - plans_executing[plan.name] = True + + if status == "Pending": continue - # Check for early completion failure - elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): - if not all_executing: - # Construct error message with status of all plans - status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) - raise AssertionError( - f"Plan {plan.name} reached {status} before all plans were executing simultaneously. " - f"Statuses: {status_msg}" - ) + if status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED) and not all_executing: + status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) + raise AssertionError( + f"Plan {plan.name} reached {status} before all plans were executing simultaneously. " + f"Statuses: {status_msg}" + ) - # Check if all plans are executing + if not plans_executing[plan.name]: + LOGGER.info(f"Plan '{plan.name}' is now executing") + plans_executing[plan.name] = True + if all(plans_executing.values()): - if not all_executing: - LOGGER.info("SUCCESS: All migrations are executing simultaneously") - all_executing = True + LOGGER.info("SUCCESS: All migrations are executing simultaneously") + all_executing = True break if not all_executing: raise AssertionError("Failed to validate all migrations executing simultaneously within timeout") Suggestion importance[1-10]: 9 __ Why: The suggestion correctly identifies a critical bug where the Plan object's status is not refreshed within the polling loop, and it also fixes a potential AttributeError when accessing plan.instance.status.conditions before it is populated.	High
	Ensure VM names are unique In the test_concurrent_xcopy_vddk_migration configuration, assign unique names to the two virtual_machines to prevent conflicts during testing, as they currently share the same name. tests/tests_config/config.py [473-492] "test_concurrent_xcopy_vddk_migration": { "virtual_machines": [ { - "name": "xcopy-template-test", + "name": "xcopy-template-test-1", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, { - "name": "xcopy-template-test", + "name": "xcopy-template-test-2", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, ], "warm_migration": False, "copyoffload": True, }, Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies that using the same VM name for two different migration plans will cause conflicts, as the test framework relies on unique names for cloning, inventory lookup, and validation.	Medium
best practice	✅ Prevent indexing errors in assertions Suggestion Impact: The assertion was adjusted to access offloadPlugin via .get("offloadPlugin") instead of attribute access, but the commit did not add the suggested non-empty map guard and still indexes spec.map[0]. code diff: assert self.storage_map_xcopy, "StorageMap creation failed for XCOPY plan" - assert self.storage_map_xcopy.instance.spec.map[0].offloadPlugin, ( + assert self.storage_map_xcopy.instance.spec.map[0].get("offloadPlugin"), ( "XCOPY StorageMap missing offloadPlugin configuration" ) In test_create_storagemap_xcopy, add a check to ensure self.storage_map_xcopy.instance.spec.map is not empty before accessing its elements to prevent potential IndexError exceptions. tests/test_copyoffload_migration.py [3269-3272] assert self.storage_map_xcopy, "StorageMap creation failed for XCOPY plan" -assert self.storage_map_xcopy.instance.spec.map[0].get("offloadPlugin"), ( - "XCOPY StorageMap missing offloadPlugin configuration" -) +mappings = (self.storage_map_xcopy.instance.spec or {}).get("map") or [] +assert mappings, "XCOPY StorageMap has empty mapping list" +assert any(m.get("offloadPlugin") for m in mappings), "XCOPY StorageMap missing offloadPlugin configuration" Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies a potential IndexError if spec.map is empty and provides a more robust way to check for the presence of offloadPlugin, improving test reliability and failure diagnostics.	Low
Possible issue	Use unique plan identifiers Use a composite key of plan.namespace and plan.name for dictionaries in wait_for_concurrent_migration_execution to uniquely identify plans, as names alone can collide across different namespaces. [utilities/mtv_migration.py [565-580]](#289...

[prabinovRedhat]

Both tests passed:

[qodo-code-review]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Type Mismatch StorageMap spec map entries are accessed inconsistently (attribute-style vs dict-style). If the underlying objects are not plain dicts, using get() may raise at runtime and make the test flaky or fail immediately. Align access style and make the assertion resilient while still producing a clear failure message when offloadPlugin is present/absent unexpectedly. assert self.storage_map_xcopy, "StorageMap creation failed for XCOPY plan" assert self.storage_map_xcopy.instance.spec.map[0].offloadPlugin, ( "XCOPY StorageMap missing offloadPlugin configuration" ) def test_create_storagemap_vddk( self, prepared_plan_2: dict[str, Any], fixture_store: dict[str, Any], ocp_admin_client: "DynamicClient", source_provider: BaseProvider, destination_provider: OCPProvider, source_provider_inventory: ForkliftInventory, target_namespace: str, ) -> None: """Create standard StorageMap for VDDK plan.""" vms_names = [vm["name"] for vm in prepared_plan_2["virtual_machines"]] self.__class__.storage_map_vddk = get_storage_migration_map( fixture_store=fixture_store, target_namespace=target_namespace, source_provider=source_provider, destination_provider=destination_provider, ocp_admin_client=ocp_admin_client, source_provider_inventory=source_provider_inventory, vms=vms_names, ) assert self.storage_map_vddk, "StorageMap creation failed for VDDK plan" assert not self.storage_map_vddk.instance.spec.map[0].get("offloadPlugin"), ( "VDDK StorageMap should NOT have offloadPlugin configuration" ) Defensive Checks Status polling assumes status and conditions are always present and iterable. In early plan lifecycle or transient API states, status/conditions may be missing or None, which would raise and abort the wait prematurely. Add defensive handling (AttributeError/TypeError/KeyError) and treat missing data as not-yet-executing to keep polling stable. def wait_for_concurrent_migration_execution(plan_list: list[Plan], timeout: int = 120) -> None: """Wait for multiple migration plans to be executing simultaneously. Polls the status of all provided plans and validates that they all reach the "Executing" state at the same time. If any plan completes (Succeeded or Failed) before all plans are executing, the validation fails. Args: plan_list: List of Plan resources to monitor. timeout: Timeout in seconds to wait for simultaneous execution. Raises: AssertionError: If plans do not execute simultaneously or if any plan completes early. """ LOGGER.info(f"Validating {len(plan_list)} migrations enter executing state simultaneously") plans_executing = {plan.name: False for plan in plan_list} all_executing = False def _check_plan_status(plan: Plan) -> str: for cond in plan.instance.status.conditions: if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: cond_type = cond["type"] if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): return cond_type return "Executing" for sample in TimeoutSampler( func=lambda: {plan.name: _check_plan_status(plan) for plan in plan_list}, sleep=2, wait_timeout=timeout, ): current_statuses = sample # Update executing state for each plan for plan in plan_list: status = current_statuses[plan.name] if status == "Executing": if not plans_executing[plan.name]: LOGGER.info(f"Plan '{plan.name}' is now executing") plans_executing[plan.name] = True # Check for early completion failure elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): if not all_executing: # Construct error message with status of all plans status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) raise AssertionError( f"Plan {plan.name} reached {status} before all plans were executing simultaneously. " f"Statuses: {status_msg}" ) # Check if all plans are executing if all(plans_executing.values()): if not all_executing: LOGGER.info("SUCCESS: All migrations are executing simultaneously") all_executing = True break if not all_executing: raise AssertionError("Failed to validate all migrations executing simultaneously within timeout") Test Data The new concurrent test config defines two VMs with the same name and sets copyoffload true globally, which risks name collisions and makes it unclear how the VDDK plan is differentiated from the XCOPY plan. Ensure the config clearly provides distinct inputs for XCOPY vs VDDK (unique VM identities and/or per-plan copyoffload settings) so the test validates what it claims. "test_concurrent_xcopy_vddk_migration": { "virtual_machines": [ { "name": "xcopy-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, { "name": "xcopy-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, ], "warm_migration": False, "copyoffload": True, },

---

💡 Tool usage guide:

---

Overview:
The review tool scans the PR code changes, and generates a PR review which includes several types of feedbacks, such as possible PR issues, security threats and relevant test in the PR. More feedbacks can be added by configuring the tool.

The tool can be triggered automatically every time a new PR is opened, or can be invoked manually by commenting on any PR.

• When commenting, to edit configurations related to the review tool (pr_reviewer section), use the following template:

/review --pr_reviewer.some_config1=... --pr_reviewer.some_config2=...

• With a configuration file, use the following template:

[pr_reviewer]
some_config1=...
some_config2=...

See the review usage page for a comprehensive guide on using this tool.

[prabinovRedhat]

/verified

[coderabbitai]

Actionable comments posted: 4

🤖 Fix all issues with AI agents

In `@tests/test_copyoffload_migration.py`:
- Around line 3255-3258: The assertion currently accesses offloadPlugin via
attribute access on self.storage_map_xcopy.instance.spec.map[0].offloadPlugin
which can raise AttributeError; change it to use dict-style access/ResourceField
get (e.g., self.storage_map_xcopy.instance.spec.map[0].get('offloadPlugin')) so
the assertion fails with your custom message if the key is missing (follow the
same pattern used in the VDDK check below), keeping the same assertion text.
- Around line 3214-3227: Update the one-line docstring of the
test_create_storagemap_xcopy function to the project's Args/Returns/Raises
format: expand the docstring to describe the purpose briefly and add an "Args"
section listing each fixture parameter (prepared_plan_1, fixture_store,
ocp_admin_client, source_provider, destination_provider,
source_provider_inventory, target_namespace, source_provider_data,
copyoffload_storage_secret) with their types, a "Returns" section (e.g., None),
and a "Raises" section indicating any expected exceptions (or "None" if none
expected); keep the text concise and consistent with other tests in this class
and use the exact function name test_create_storagemap_xcopy to locate where to
update.

In `@utilities/mtv_migration.py`:
- Around line 547-560: The docstring for wait_for_concurrent_migration_execution
is missing a Returns section; update the docstring for the function
wait_for_concurrent_migration_execution to include a Returns entry that states
it returns None and briefly describes the meaning (e.g., "None: returns nothing;
raises AssertionError on failure") so the docstring follows the project's
Args/Returns/Raises format.

[qodo-code-review]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Type Mismatch The VDDK StorageMap assertion mixes attribute-style access and dict-style access on the same spec.map[0] object; if spec.map[0] is not a dict, calling .get() will raise at runtime. Use a consistent access pattern (e.g., getattr(..., "offloadPlugin", None) or inspect the underlying dict representation) so the test fails for the intended reason (unexpected field presence) rather than due to an AttributeError. def test_create_storagemap_vddk( self, prepared_plan_2: dict[str, Any], fixture_store: dict[str, Any], ocp_admin_client: "DynamicClient", source_provider: BaseProvider, destination_provider: OCPProvider, source_provider_inventory: ForkliftInventory, target_namespace: str, ) -> None: """Create standard StorageMap for VDDK plan.""" vms_names = [vm["name"] for vm in prepared_plan_2["virtual_machines"]] self.__class__.storage_map_vddk = get_storage_migration_map( fixture_store=fixture_store, target_namespace=target_namespace, source_provider=source_provider, destination_provider=destination_provider, ocp_admin_client=ocp_admin_client, source_provider_inventory=source_provider_inventory, vms=vms_names, ) assert self.storage_map_vddk, "StorageMap creation failed for VDDK plan" assert not self.storage_map_vddk.instance.spec.map[0].get("offloadPlugin"), ( "VDDK StorageMap should NOT have offloadPlugin configuration" ) False Positive The concurrent-execution validator treats the absence of a terminal advisory condition as "Executing", which can incorrectly mark plans as executing while they are still Pending/Starting/Validating. This can cause flaky tests and missed regressions; the validator should explicitly detect an "executing/in-progress" signal rather than defaulting to executing when no succeeded/failed condition is found. def wait_for_concurrent_migration_execution(plan_list: list[Plan], timeout: int = 120) -> None: """Wait for multiple migration plans to be executing simultaneously. Polls the status of all provided plans and validates that they all reach the "Executing" state at the same time. If any plan completes (Succeeded or Failed) before all plans are executing, the validation fails. Args: plan_list: List of Plan resources to monitor. timeout: Timeout in seconds to wait for simultaneous execution. Raises: AssertionError: If plans do not execute simultaneously or if any plan completes early. """ LOGGER.info(f"Validating {len(plan_list)} migrations enter executing state simultaneously") plans_executing = {plan.name: False for plan in plan_list} all_executing = False def _check_plan_status(plan: Plan) -> str: for cond in plan.instance.status.conditions: if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: cond_type = cond["type"] if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): return cond_type return "Executing" for sample in TimeoutSampler( func=lambda: {plan.name: _check_plan_status(plan) for plan in plan_list}, sleep=2, wait_timeout=timeout, ): current_statuses = sample # Update executing state for each plan for plan in plan_list: status = current_statuses[plan.name] if status == "Executing": if not plans_executing[plan.name]: LOGGER.info(f"Plan '{plan.name}' is now executing") plans_executing[plan.name] = True # Check for early completion failure elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): if not all_executing: # Construct error message with status of all plans status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) raise AssertionError( f"Plan {plan.name} reached {status} before all plans were executing simultaneously. " f"Statuses: {status_msg}" ) # Check if all plans are executing if all(plans_executing.values()): if not all_executing: LOGGER.info("SUCCESS: All migrations are executing simultaneously") all_executing = True break if not all_executing: raise AssertionError("Failed to validate all migrations executing simultaneously within timeout") Test Config The new test config lists two identical VM entries with the same name, while the test logic expects two distinct plans (XCOPY vs VDDK). This can lead to ambiguous inventory lookups, duplicate migration targets, or unintended coupling between the two plan preparations; ensure the config clearly represents two separate plan inputs (or distinct VMs) matching how prepared_plan_1 and prepared_plan_2 are derived. "test_concurrent_xcopy_vddk_migration": { "virtual_machines": [ { "name": "xcopy-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, { "name": "xcopy-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, ], "warm_migration": False, "copyoffload": True, },

---

💡 Tool usage guide:

---

Overview:
The review tool scans the PR code changes, and generates a PR review which includes several types of feedbacks, such as possible PR issues, security threats and relevant test in the PR. More feedbacks can be added by configuring the tool.

The tool can be triggered automatically every time a new PR is opened, or can be invoked manually by commenting on any PR.

• When commenting, to edit configurations related to the review tool (pr_reviewer section), use the following template:

/review --pr_reviewer.some_config1=... --pr_reviewer.some_config2=...

• With a configuration file, use the following template:

[pr_reviewer]
some_config1=...
some_config2=...

See the review usage page for a comprehensive guide on using this tool.

[qodo-code-review]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 4 🔵🔵🔵🔵⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Possible Issue The new plan-status detection mixes condition-based completion detection with a fallback that infers execution by merely finding a Migration CR. This can misclassify a plan as executing even when it is pending/queued, and it returns an empty string for “unknown” which may cause the polling loops to behave unpredictably. Validate that the returned values are consistent (use Plan.Status constants vs free-form strings) and that execution detection is based on authoritative status fields rather than existence of related resources. def get_plan_migration_status(plan: Plan) -> str: """Get the migration status from the Plan conditions. Args: plan (Plan): The Plan resource to check. Returns: str: The status of the plan ("Executing", "Succeeded", "Failed", or empty string if unknown). Raises: None """ status = plan.instance.status if not status: return "" conditions = status.conditions if conditions: for cond in conditions: if cond["category"] == "Advisory" and cond["status"] == Plan.Condition.Status.TRUE: cond_type = cond["type"] if cond_type in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): return cond_type # Check if Migration CR exists to confirm execution try: _find_migration_for_plan(plan) return Plan.Status.EXECUTING except MigrationNotFoundError: return "" def wait_for_migration_complate(plan: Plan) -> None: try: last_status: str = "" for sample in TimeoutSampler( func=get_plan_migration_status, sleep=1, wait_timeout=py_config.get("plan_wait_timeout", 600), plan=plan, ): if sample != last_status: LOGGER.info(f"Plan '{plan.name}' migration status: '{sample}'") Flaky Test Risk Concurrent execution validation polls all plans and fails on early completion, but it relies on per-plan status reads without an explicit refresh/rehydration step and assumes plan names are unique keys. Confirm the Plan objects are refreshed on each poll (otherwise you may read stale status), and consider keying by namespace+name to avoid collisions in shared namespaces. def wait_for_concurrent_migration_execution(plan_list: list[Plan], timeout: int = 120) -> None: """Wait for multiple migration plans to be executing simultaneously. Polls the status of all provided plans and validates that they all reach the "Executing" state at the same time. If any plan completes (Succeeded or Failed) before all plans are executing, the validation fails. Args: plan_list: List of Plan resources to monitor. timeout: Timeout in seconds to wait for simultaneous execution. Returns: None Raises: AssertionError: If plans do not execute simultaneously or if any plan completes early. """ LOGGER.info(f"Validating {len(plan_list)} migrations enter executing state simultaneously") plans_executing = {plan.name: False for plan in plan_list} try: for current_statuses in TimeoutSampler( func=lambda: {plan.name: get_plan_migration_status(plan) for plan in plan_list}, sleep=2, wait_timeout=timeout, ): # Update executing state for each plan for plan in plan_list: status = current_statuses[plan.name] if status == Plan.Status.EXECUTING: if not plans_executing[plan.name]: LOGGER.info(f"Plan '{plan.name}' is now executing") plans_executing[plan.name] = True continue # Check for early completion failure elif status in (Plan.Status.SUCCEEDED, Plan.Status.FAILED): # Construct error message with status of all plans status_msg = ", ".join([f"{name}: {stat}" for name, stat in current_statuses.items()]) raise AssertionError( f"Plan {plan.name} reached {status} before all plans were executing simultaneously. " f"Statuses: {status_msg}" ) # Check if all plans are executing if all(plans_executing.values()): LOGGER.info("SUCCESS: All migrations are executing simultaneously") return Test Data Issue The new configuration defines two virtual machines with the same name, which can lead to collisions (clone name conflicts, inventory lookups mapping to the same VM, or unexpected reuse) and can undermine the intent of having distinct XCOPY vs VDDK plans. Ensure the two entries are distinguishable and that the config fields align with how the fixtures split plan 1 vs plan 2 parameters (especially around copyoffload behavior). "test_concurrent_xcopy_vddk_migration": { "virtual_machines": [ { "name": "xcopy-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, { "name": "xcopy-template-test", "source_vm_power": "off", "guest_agent": True, "clone": True, "disk_type": "thick-lazy", }, ], "warm_migration": False, "copyoffload": True, },

---

💡 Tool usage guide:

---

Overview:
The review tool scans the PR code changes, and generates a PR review which includes several types of feedbacks, such as possible PR issues, security threats and relevant test in the PR. More feedbacks can be added by configuring the tool.

The tool can be triggered automatically every time a new PR is opened, or can be invoked manually by commenting on any PR.

• When commenting, to edit configurations related to the review tool (pr_reviewer section), use the following template:

/review --pr_reviewer.some_config1=... --pr_reviewer.some_config2=...

• With a configuration file, use the following template:

[pr_reviewer]
some_config1=...
some_config2=...

See the review usage page for a comprehensive guide on using this tool.

[prabinovRedhat]

/verified

[myakove]

/approve

[redhat-qe-bot]

Failed to build and push ghcr.io/redhatqe/mtv-api-tests:latest