From b7c74c6c0e3ded5175aa692f1c25e0779d9233ae Mon Sep 17 00:00:00 2001 From: Christopher Hart Date: Fri, 6 Feb 2026 08:36:49 -0500 Subject: [PATCH 1/4] fix: defer project issue creation until catalog PR exists for catalog-destined test cases For catalog-destined test cases, project issues should only be created after a catalog PR exists so the issue can reference it. This prevents orphaned issues from being created when test cases have no catalog PR yet. Changes: - requires_issue_creation now checks metadata.catalog.destined and requires metadata.catalog_tracking.pr_number before allowing issue creation for catalog-destined test cases - Reorder process_test_requirements loop: catalog PR creation runs before issue creation so both happen in the same CI run - Add unit tests for new catalog-destined gating behavior - Add integration tests for single-run catalog PR + issue creation Co-Authored-By: Claude Opus 4.6 --- .../processing/test_cases_processor.py | 23 ++++- .../synchronize/test_requirements.py | 52 +++++----- .../test_processing_test_cases_processor.py | 41 +++++++- .../test_synchronize_test_requirements.py | 96 +++++++++++++++++++ 4 files changed, 183 insertions(+), 29 deletions(-) diff --git a/github_ops_manager/processing/test_cases_processor.py b/github_ops_manager/processing/test_cases_processor.py index b667bdf..6a90433 100644 --- a/github_ops_manager/processing/test_cases_processor.py +++ b/github_ops_manager/processing/test_cases_processor.py @@ -352,8 +352,12 @@ def update_test_case_with_project_pr_metadata( def requires_issue_creation(test_case: dict[str, Any]) -> bool: """Check if a test case needs an issue to be created. - An issue is needed if the test case doesn't already have issue metadata. - Checks nested structure: metadata.project_tracking.{issue_number, issue_url} + An issue is needed if: + - The test case doesn't already have issue metadata + (metadata.project_tracking.{issue_number, issue_url}) + - For catalog-destined test cases, a catalog PR must already exist + (metadata.catalog_tracking.pr_number) before an issue is created. + This ensures the project issue can reference the catalog PR. Args: test_case: Test case dictionary to check @@ -361,11 +365,22 @@ def requires_issue_creation(test_case: dict[str, Any]) -> bool: Returns: True if issue needs to be created, False otherwise """ - project_tracking = test_case.get("metadata", {}).get("project_tracking", {}) + metadata = test_case.get("metadata", {}) + project_tracking = metadata.get("project_tracking", {}) has_issue_number = project_tracking.get("issue_number") is not None has_issue_url = project_tracking.get("issue_url") is not None - return not (has_issue_number and has_issue_url) + if has_issue_number and has_issue_url: + return False + + # For catalog-destined test cases, defer issue creation until + # a catalog PR exists so the issue can reference it + is_catalog = metadata.get("catalog", {}).get("destined", False) + if is_catalog: + has_catalog_pr = metadata.get("catalog_tracking", {}).get("pr_number") is not None + return has_catalog_pr + + return True def requires_project_pr_creation(test_case: dict[str, Any]) -> bool: diff --git a/github_ops_manager/synchronize/test_requirements.py b/github_ops_manager/synchronize/test_requirements.py index 11370b7..df7a61a 100644 --- a/github_ops_manager/synchronize/test_requirements.py +++ b/github_ops_manager/synchronize/test_requirements.py @@ -444,7 +444,34 @@ async def process_test_requirements( title = test_case.get("title", "Unknown") logger.info("Processing test case", title=title) - # Check if issue needs to be created + # Create catalog PR first (if needed) so that catalog-destined test + # cases have catalog_tracking metadata before issue creation runs. + # This allows the project issue to reference the catalog PR. + if requires_catalog_pr_creation(test_case): + if not catalog_adapter or not catalog_default_branch or not catalog_repo_url: + logger.warning( + "Catalog PR needed but catalog configuration not provided", + title=title, + ) + results["errors"].append(f"Catalog PR needed for {title} but catalog not configured") + continue + + pr_result = await create_catalog_pr_for_test_case( + test_case, + catalog_adapter, + base_directory, + catalog_default_branch, + catalog_repo_url, + ) + + if pr_result: + results["catalog_prs_created"] += 1 + # Save metadata back to file + save_test_case_metadata(test_case) + + # Check if issue needs to be created. For catalog-destined test cases, + # this will only proceed if a catalog PR already exists (either created + # above or in a previous run). if requires_issue_creation(test_case): if template: try: @@ -487,29 +514,6 @@ async def process_test_requirements( # Save metadata back to file save_test_case_metadata(test_case) - # Check if catalog PR needs to be created - if requires_catalog_pr_creation(test_case): - if not catalog_adapter or not catalog_default_branch or not catalog_repo_url: - logger.warning( - "Catalog PR needed but catalog configuration not provided", - title=title, - ) - results["errors"].append(f"Catalog PR needed for {title} but catalog not configured") - continue - - pr_result = await create_catalog_pr_for_test_case( - test_case, - catalog_adapter, - base_directory, - catalog_default_branch, - catalog_repo_url, - ) - - if pr_result: - results["catalog_prs_created"] += 1 - # Save metadata back to file - save_test_case_metadata(test_case) - logger.info( "Completed test requirements processing", total=results["total_test_cases"], diff --git a/tests/unit/test_processing_test_cases_processor.py b/tests/unit/test_processing_test_cases_processor.py index e02dba1..c01adaa 100644 --- a/tests/unit/test_processing_test_cases_processor.py +++ b/tests/unit/test_processing_test_cases_processor.py @@ -159,7 +159,7 @@ class TestRequiresIssueCreation: """Tests for requires_issue_creation function.""" def test_needs_issue_when_no_metadata(self) -> None: - """Should return True when no issue metadata exists.""" + """Should return True when no issue metadata exists (non-catalog).""" test_case: dict[str, Any] = {"title": "Test Case 1"} assert requires_issue_creation(test_case) is True @@ -192,6 +192,45 @@ def test_no_issue_needed_when_both_exist(self) -> None: } assert requires_issue_creation(test_case) is False + def test_catalog_destined_defers_without_catalog_pr(self) -> None: + """Should return False for catalog-destined test case without catalog PR.""" + test_case: dict[str, Any] = { + "title": "Test Case 1", + "metadata": {"catalog": {"destined": True}}, + } + assert requires_issue_creation(test_case) is False + + def test_catalog_destined_proceeds_with_catalog_pr(self) -> None: + """Should return True for catalog-destined test case with catalog PR.""" + test_case: dict[str, Any] = { + "title": "Test Case 1", + "metadata": { + "catalog": {"destined": True}, + "catalog_tracking": {"pr_number": 456, "pr_url": "https://catalog-pr"}, + }, + } + assert requires_issue_creation(test_case) is True + + def test_catalog_destined_skips_when_issue_already_exists(self) -> None: + """Should return False for catalog-destined test case that already has an issue.""" + test_case: dict[str, Any] = { + "title": "Test Case 1", + "metadata": { + "catalog": {"destined": True}, + "catalog_tracking": {"pr_number": 456, "pr_url": "https://catalog-pr"}, + "project_tracking": {"issue_number": 10, "issue_url": "https://issue"}, + }, + } + assert requires_issue_creation(test_case) is False + + def test_non_catalog_destined_ignores_catalog_tracking(self) -> None: + """Should return True for non-catalog test case regardless of catalog_tracking.""" + test_case: dict[str, Any] = { + "title": "Test Case 1", + "metadata": {"catalog": {"destined": False}}, + } + assert requires_issue_creation(test_case) is True + class TestRequiresProjectPrCreation: """Tests for requires_project_pr_creation function.""" diff --git a/tests/unit/test_synchronize_test_requirements.py b/tests/unit/test_synchronize_test_requirements.py index 21bd871..9c9c7f3 100644 --- a/tests/unit/test_synchronize_test_requirements.py +++ b/tests/unit/test_synchronize_test_requirements.py @@ -611,6 +611,102 @@ async def test_creates_catalog_prs_for_catalog_destined(self) -> None: assert results["catalog_prs_created"] == 1 mock_catalog_adapter.create_pull_request.assert_called_once() + @pytest.mark.asyncio + async def test_catalog_destined_creates_issue_after_catalog_pr(self) -> None: + """Should create both catalog PR and issue in one run for catalog-destined test case.""" + with tempfile.TemporaryDirectory() as tmpdir: + test_cases_dir = Path(tmpdir) + script_path = "verify_nxos_test.robot" + robot_content = """*** Settings *** +Test Tags os:nxos + +*** Test Cases *** +Test + Log Hello +""" + (test_cases_dir / script_path).write_text(robot_content) + + (test_cases_dir / "my_test_cases.yaml").write_text( + f"""test_cases: + - title: Test Case 1 + purpose: Test purpose + generated_script_path: {script_path} + metadata: + catalog: + destined: true + commands: + - command: show version +""" + ) + + mock_project_adapter = AsyncMock() + mock_issue = MagicMock() + mock_issue.number = 50 + mock_issue.html_url = "https://github.com/org/repo/issues/50" + mock_project_adapter.create_issue.return_value = mock_issue + + mock_catalog_adapter = AsyncMock() + mock_catalog_adapter.branch_exists.return_value = False + mock_pr = MagicMock() + mock_pr.number = 20 + mock_pr.html_url = "https://github.com/catalog/repo/pull/20" + mock_pr.head.ref = "feat/nxos/add-test" + mock_catalog_adapter.create_pull_request.return_value = mock_pr + + with patch( + "github_ops_manager.synchronize.test_requirements.save_test_case_metadata", + return_value=True, + ): + results = await process_test_requirements( + test_cases_dir=test_cases_dir, + base_directory=test_cases_dir, + project_adapter=mock_project_adapter, + project_default_branch="main", + project_repo_url="https://github.com/org/repo", + catalog_adapter=mock_catalog_adapter, + catalog_default_branch="main", + catalog_repo_url="https://github.com/catalog/repo", + ) + + # Catalog PR created first, then issue + assert results["catalog_prs_created"] == 1 + assert results["issues_created"] == 1 + mock_catalog_adapter.create_pull_request.assert_called_once() + mock_project_adapter.create_issue.assert_called_once() + + @pytest.mark.asyncio + async def test_catalog_destined_skips_issue_without_catalog_pr(self) -> None: + """Should NOT create issue for catalog-destined test case without generated script.""" + with tempfile.TemporaryDirectory() as tmpdir: + test_cases_dir = Path(tmpdir) + + # No generated_script_path, so no catalog PR will be created + (test_cases_dir / "my_test_cases.yaml").write_text( + """test_cases: + - title: Test Case 1 + purpose: Test purpose + metadata: + catalog: + destined: true + commands: + - command: show version +""" + ) + + mock_project_adapter = AsyncMock() + + results = await process_test_requirements( + test_cases_dir=test_cases_dir, + base_directory=test_cases_dir, + project_adapter=mock_project_adapter, + project_default_branch="main", + project_repo_url="https://github.com/org/repo", + ) + + assert results["issues_created"] == 0 + assert results["catalog_prs_created"] == 0 + mock_project_adapter.create_issue.assert_not_called() + @pytest.mark.asyncio async def test_reports_error_when_catalog_not_configured(self) -> None: """Should report error when catalog PR needed but not configured.""" From 0e3d0f34320365c7e80c68b5d960a4a6a5209830 Mon Sep 17 00:00:00 2001 From: Christopher Hart Date: Fri, 6 Feb 2026 09:01:52 -0500 Subject: [PATCH 2/4] feat: use split-style tracking issue template for catalog-destined test cases For catalog-destined test cases, process_test_requirements now creates tracking issues using tracking_issue.j2 (split style with catalog PR reference and task checklist) instead of tac_issues_body.j2 (collapsed style with test requirement details only). Non-catalog test cases continue to use the collapsed template. Adds create_tracking_issue_for_catalog_test_case() which renders the tracking template from test case metadata, supporting both same-run catalog PRs and catalog PRs from previous runs. Co-Authored-By: Claude Opus 4.6 --- .../synchronize/test_requirements.py | 213 +++++++++-- .../test_synchronize_test_requirements.py | 339 +++++++++++++++++- 2 files changed, 528 insertions(+), 24 deletions(-) diff --git a/github_ops_manager/synchronize/test_requirements.py b/github_ops_manager/synchronize/test_requirements.py index df7a61a..42f21f1 100644 --- a/github_ops_manager/synchronize/test_requirements.py +++ b/github_ops_manager/synchronize/test_requirements.py @@ -25,6 +25,11 @@ update_test_case_with_pr_metadata, update_test_case_with_project_pr_metadata, ) +from github_ops_manager.synchronize.tracking_issues import ( + compute_project_branch_name, + load_tracking_issue_template, + strip_os_tag_from_title, +) from github_ops_manager.utils.constants import DEFAULT_MAX_ISSUE_BODY_LENGTH from github_ops_manager.utils.templates import construct_jinja2_template_from_file from github_ops_manager.utils.truncation import truncate_data_dict_outputs @@ -84,6 +89,157 @@ async def create_issue_for_test_case( return None +def _extract_os_from_catalog_branch(branch: str) -> str | None: + """Extract OS name from catalog branch name pattern. + + Catalog branches follow the pattern: feat/{os_name}/add-{stem} + + Args: + branch: Catalog branch name (e.g., "feat/nxos/add-verify-nxos-interfaces") + + Returns: + OS name (e.g., "nxos") or None if pattern doesn't match + """ + parts = branch.split("/") + if len(parts) >= 3 and parts[0] in ("feat", "feature"): + return parts[1] + return None + + +async def create_tracking_issue_for_catalog_test_case( + test_case: dict[str, Any], + project_adapter: GitHubKitAdapter, + catalog_repo_url: str, + labels: list[str] | None = None, + catalog_pr_result: dict[str, Any] | None = None, +) -> dict[str, Any] | None: + """Create a split-style tracking issue for a catalog-destined test case. + + Uses the tracking_issue.j2 template to create an issue that references + the catalog PR and provides a task checklist for parameter learning. + + This works for both same-run catalog PRs (catalog_pr_result provided) + and previous-run catalog PRs (metadata already in test case). + + Args: + test_case: Test case dictionary with catalog_tracking metadata + project_adapter: GitHub adapter for project repository + catalog_repo_url: Full URL to catalog repository + labels: Optional list of labels to apply + catalog_pr_result: Optional result from create_catalog_pr_for_test_case + (provides os_name when catalog PR was created in same run) + + Returns: + Created issue data or None on error + """ + title = test_case.get("title") + if not title: + logger.error("Test case missing title, cannot create tracking issue") + return None + + # Get catalog PR metadata from test case + catalog_tracking = test_case.get("metadata", {}).get("catalog_tracking", {}) + catalog_pr_number = catalog_tracking.get("pr_number") + catalog_pr_url = catalog_tracking.get("pr_url") + catalog_branch = catalog_tracking.get("pr_branch") + + if not catalog_pr_number or not catalog_pr_url or not catalog_branch: + logger.error( + "Test case missing catalog PR metadata for tracking issue", + title=title, + ) + return None + + # Extract OS name: prefer same-run result, fall back to branch parsing + os_name = None + if catalog_pr_result: + os_name = catalog_pr_result.get("os_name") + if not os_name: + os_name = _extract_os_from_catalog_branch(catalog_branch) + if not os_name: + logger.error("Could not determine OS name for tracking issue", title=title) + return None + + # Construct catalog PR title from known pattern + catalog_dir = normalize_os_to_catalog_dir(os_name) + catalog_pr_title = f"feat: add {catalog_dir} test - {title}" + + # Use helpers from tracking_issues module + clean_title = strip_os_tag_from_title(title) + suggested_branch = compute_project_branch_name(catalog_branch) + + # Build test requirement data from test case + commands_list = [] + if "commands" in test_case: + for cmd in test_case["commands"]: + if isinstance(cmd, dict): + commands_list.append(cmd.get("command", "")) + else: + commands_list.append(str(cmd)) + + test_requirement = { + "purpose": test_case.get("purpose", ""), + "commands": commands_list, + "pass_criteria": test_case.get("pass_criteria", ""), + "sample_parameters": test_case.get("jobfile_parameters", ""), + "parameters_to_parsed_data_mapping": test_case.get("jobfile_parameters_mapping", ""), + } + + # Load and render the tracking issue template + template = load_tracking_issue_template() + issue_body = template.render( + catalog_pr_title=catalog_pr_title, + catalog_pr_url=catalog_pr_url, + catalog_pr_number=catalog_pr_number, + catalog_branch=catalog_branch, + suggested_project_branch=suggested_branch, + test_case_title=title, + test_case_title_clean=clean_title, + os_name=os_name.upper(), + test_requirement=test_requirement, + ) + + issue_title = f"Review Catalog PR and Learn Parameters: {title}" + + logger.info( + "Creating tracking issue for catalog-destined test case", + title=title, + catalog_pr_number=catalog_pr_number, + ) + + try: + issue = await project_adapter.create_issue( + title=issue_title, + body=issue_body, + labels=labels, + ) + + logger.info( + "Created tracking issue for catalog-destined test case", + title=title, + issue_number=issue.number, + issue_url=issue.html_url, + catalog_pr_number=catalog_pr_number, + ) + + # Update test case with issue metadata + update_test_case_with_issue_metadata(test_case, issue.number, issue.html_url) + + return { + "issue": issue, + "issue_number": issue.number, + "issue_url": issue.html_url, + } + + except Exception as e: + logger.error( + "Failed to create tracking issue for catalog-destined test case", + title=title, + error=str(e), + ) + return None + + async def create_project_pr_for_test_case( test_case: dict[str, Any], github_adapter: GitHubKitAdapter, @@ -444,6 +600,9 @@ async def process_test_requirements( title = test_case.get("title", "Unknown") logger.info("Processing test case", title=title) + # Track catalog PR result for use during issue creation (same-run) + catalog_pr_result = None + # Create catalog PR first (if needed) so that catalog-destined test # cases have catalog_tracking metadata before issue creation runs. # This allows the project issue to reference the catalog PR. @@ -456,7 +615,7 @@ async def process_test_requirements( results["errors"].append(f"Catalog PR needed for {title} but catalog not configured") continue - pr_result = await create_catalog_pr_for_test_case( + catalog_pr_result = await create_catalog_pr_for_test_case( test_case, catalog_adapter, base_directory, @@ -464,7 +623,7 @@ async def process_test_requirements( catalog_repo_url, ) - if pr_result: + if catalog_pr_result: results["catalog_prs_created"] += 1 # Save metadata back to file save_test_case_metadata(test_case) @@ -473,26 +632,40 @@ async def process_test_requirements( # this will only proceed if a catalog PR already exists (either created # above or in a previous run). if requires_issue_creation(test_case): - if template: - try: - issue_body = render_issue_body_for_test_case(test_case, template, max_body_length=max_body_length) - except Exception as e: - logger.error("Failed to render issue body", title=title, error=str(e)) - results["errors"].append(f"Failed to render issue body for {title}: {e}") - continue - else: - # Simple default body - issue_body = f"Test requirement: {title}\n\n{test_case.get('purpose', '')}" - - # Get labels from test case or use default + is_catalog = test_case.get("metadata", {}).get("catalog", {}).get("destined", False) labels = test_case.get("labels", issue_labels) - issue_result = await create_issue_for_test_case( - test_case, - project_adapter, - issue_body, - labels=labels, - ) + if is_catalog: + # Use split-style tracking issue for catalog-destined test cases. + # This renders tracking_issue.j2 with catalog PR reference and + # task checklist for parameter learning workflow. + issue_result = await create_tracking_issue_for_catalog_test_case( + test_case, + project_adapter, + catalog_repo_url or "", + labels=labels, + catalog_pr_result=catalog_pr_result, + ) + else: + # Use collapsed-style issue for non-catalog test cases. + # This renders the issue template with full test requirement details. + if template: + try: + issue_body = render_issue_body_for_test_case(test_case, template, max_body_length=max_body_length) + except Exception as e: + logger.error("Failed to render issue body", title=title, error=str(e)) + results["errors"].append(f"Failed to render issue body for {title}: {e}") + continue + else: + # Simple default body + issue_body = f"Test requirement: {title}\n\n{test_case.get('purpose', '')}" + + issue_result = await create_issue_for_test_case( + test_case, + project_adapter, + issue_body, + labels=labels, + ) if issue_result: results["issues_created"] += 1 diff --git a/tests/unit/test_synchronize_test_requirements.py b/tests/unit/test_synchronize_test_requirements.py index 9c9c7f3..1e66252 100644 --- a/tests/unit/test_synchronize_test_requirements.py +++ b/tests/unit/test_synchronize_test_requirements.py @@ -9,9 +9,11 @@ import pytest from github_ops_manager.synchronize.test_requirements import ( + _extract_os_from_catalog_branch, create_catalog_pr_for_test_case, create_issue_for_test_case, create_project_pr_for_test_case, + create_tracking_issue_for_catalog_test_case, process_test_requirements, render_issue_body_for_test_case, ) @@ -436,6 +438,236 @@ async def test_returns_none_when_os_not_detected(self) -> None: assert result is None +class TestExtractOsFromCatalogBranch: + """Tests for _extract_os_from_catalog_branch helper.""" + + def test_extracts_os_from_feat_branch(self) -> None: + """Should extract OS from feat/{os}/add-{stem} pattern.""" + assert _extract_os_from_catalog_branch("feat/nxos/add-verify-nxos-interfaces") == "nxos" + + def test_extracts_os_from_feature_branch(self) -> None: + """Should extract OS from feature/{os}/add-{stem} pattern.""" + assert _extract_os_from_catalog_branch("feature/ios-xe/add-verify-interfaces") == "ios-xe" + + def test_returns_none_for_no_match(self) -> None: + """Should return None for branches that don't match the pattern.""" + assert _extract_os_from_catalog_branch("main") is None + assert _extract_os_from_catalog_branch("fix/something") is None + + def test_handles_deeply_nested_branch(self) -> None: + """Should extract OS from branches with more than 3 segments.""" + assert _extract_os_from_catalog_branch("feat/ios-xr/add-verify-bgp-neighbors") == "ios-xr" + + +class TestCreateTrackingIssueForCatalogTestCase: + """Tests for create_tracking_issue_for_catalog_test_case function.""" + + @pytest.mark.asyncio + async def test_creates_tracking_issue_from_metadata(self) -> None: + """Should create a tracking issue using catalog_tracking metadata.""" + mock_adapter = AsyncMock() + mock_issue = MagicMock() + mock_issue.number = 42 + mock_issue.html_url = "https://github.com/org/repo/issues/42" + mock_adapter.create_issue.return_value = mock_issue + + test_case: dict[str, Any] = { + "title": "[NX-OS] Verify Interface Status", + "purpose": "Check all interfaces are up", + "commands": [{"command": "show interface status"}], + "metadata": { + "catalog": {"destined": True}, + "catalog_tracking": { + "pr_number": 101, + "pr_url": "https://github.com/catalog/repo/pull/101", + "pr_branch": "feat/nxos/add-verify-nxos-interface-status", + "git_url": "https://github.com/catalog/repo", + }, + }, + } + + result = await create_tracking_issue_for_catalog_test_case( + test_case, + mock_adapter, + "https://github.com/catalog/repo", + ) + + assert result is not None + assert result["issue_number"] == 42 + assert result["issue_url"] == "https://github.com/org/repo/issues/42" + + # Verify issue title uses tracking issue format + call_kwargs = mock_adapter.create_issue.call_args[1] + assert call_kwargs["title"] == "Review Catalog PR and Learn Parameters: [NX-OS] Verify Interface Status" + + # Verify body contains tracking issue template elements + body = call_kwargs["body"] + assert "Catalog PR" in body + assert "https://github.com/catalog/repo/pull/101" in body + assert "feat/nxos/add-verify-nxos-interface-status" in body + assert "Tasks" in body + + @pytest.mark.asyncio + async def test_uses_os_from_catalog_pr_result(self) -> None: + """Should prefer os_name from catalog_pr_result when available.""" + mock_adapter = AsyncMock() + mock_issue = MagicMock() + mock_issue.number = 1 + mock_issue.html_url = "https://url" + mock_adapter.create_issue.return_value = mock_issue + + test_case: dict[str, Any] = { + "title": "Test Case", + "metadata": { + "catalog_tracking": { + "pr_number": 10, + "pr_url": "https://pr-url", + "pr_branch": "feat/nxos/add-test", + }, + }, + } + + catalog_pr_result = {"os_name": "nxos", "pr_number": 10} + + result = await create_tracking_issue_for_catalog_test_case( + test_case, + mock_adapter, + "https://catalog-repo", + catalog_pr_result=catalog_pr_result, + ) + + assert result is not None + body = mock_adapter.create_issue.call_args[1]["body"] + assert "NXOS" in body # os_name.upper() in template + + @pytest.mark.asyncio + async def test_updates_test_case_with_issue_metadata(self) -> None: + """Should update test case dict with issue metadata in nested structure.""" + mock_adapter = AsyncMock() + mock_issue = MagicMock() + mock_issue.number = 99 + mock_issue.html_url = "https://github.com/org/repo/issues/99" + mock_adapter.create_issue.return_value = mock_issue + + test_case: dict[str, Any] = { + "title": "Test Case", + "metadata": { + "catalog_tracking": { + "pr_number": 10, + "pr_url": "https://pr-url", + "pr_branch": "feat/nxos/add-test", + }, + }, + } + + await create_tracking_issue_for_catalog_test_case( + test_case, + mock_adapter, + "https://catalog-repo", + ) + + assert test_case["metadata"]["project_tracking"]["issue_number"] == 99 + assert test_case["metadata"]["project_tracking"]["issue_url"] == "https://github.com/org/repo/issues/99" + + @pytest.mark.asyncio + async def test_returns_none_on_missing_title(self) -> None: + """Should return None if test case has no title.""" + mock_adapter = AsyncMock() + test_case: dict[str, Any] = {"metadata": {"catalog_tracking": {"pr_number": 1}}} + + result = await create_tracking_issue_for_catalog_test_case(test_case, mock_adapter, "https://url") + + assert result is None + mock_adapter.create_issue.assert_not_called() + + @pytest.mark.asyncio + async def test_returns_none_on_missing_catalog_metadata(self) -> None: + """Should return None if catalog_tracking metadata is missing.""" + mock_adapter = AsyncMock() + test_case: dict[str, Any] = {"title": "Test Case"} + + result = await create_tracking_issue_for_catalog_test_case(test_case, mock_adapter, "https://url") + + assert result is None + mock_adapter.create_issue.assert_not_called() + + @pytest.mark.asyncio + async def test_returns_none_on_api_error(self) -> None: + """Should return None if API call fails.""" + mock_adapter = AsyncMock() + mock_adapter.create_issue.side_effect = Exception("API Error") + + test_case: dict[str, Any] = { + "title": "Test Case", + "metadata": { + "catalog_tracking": { + "pr_number": 10, + "pr_url": "https://pr-url", + "pr_branch": "feat/nxos/add-test", + }, + }, + } + + result = await create_tracking_issue_for_catalog_test_case(test_case, mock_adapter, "https://url") + + assert result is None + + @pytest.mark.asyncio + async def test_applies_labels(self) -> None: + """Should pass labels to create_issue.""" + mock_adapter = AsyncMock() + mock_issue = MagicMock() + mock_issue.number = 1 + mock_issue.html_url = "https://url" + mock_adapter.create_issue.return_value = mock_issue + + test_case: dict[str, Any] = { + "title": "Test Case", + "metadata": { + "catalog_tracking": { + "pr_number": 10, + "pr_url": "https://pr-url", + "pr_branch": "feat/nxos/add-test", + }, + }, + } + + await create_tracking_issue_for_catalog_test_case( + test_case, + mock_adapter, + "https://url", + labels=["test-automation", "catalog"], + ) + + call_kwargs = mock_adapter.create_issue.call_args[1] + assert call_kwargs["labels"] == ["test-automation", "catalog"] + + @pytest.mark.asyncio + async def test_renders_suggested_branch_name(self) -> None: + """Should render suggested project branch name from catalog branch.""" + mock_adapter = AsyncMock() + mock_issue = MagicMock() + mock_issue.number = 1 + mock_issue.html_url = "https://url" + mock_adapter.create_issue.return_value = mock_issue + + test_case: dict[str, Any] = { + "title": "Test Case", + "metadata": { + "catalog_tracking": { + "pr_number": 10, + "pr_url": "https://pr-url", + "pr_branch": "feat/nxos/add-verify-nxos-interfaces", + }, + }, + } + + await create_tracking_issue_for_catalog_test_case(test_case, mock_adapter, "https://url") + + body = mock_adapter.create_issue.call_args[1]["body"] + assert "learn/nxos/add-verify-nxos-interfaces" in body + + class TestProcessTestRequirements: """Tests for process_test_requirements function.""" @@ -612,8 +844,8 @@ async def test_creates_catalog_prs_for_catalog_destined(self) -> None: mock_catalog_adapter.create_pull_request.assert_called_once() @pytest.mark.asyncio - async def test_catalog_destined_creates_issue_after_catalog_pr(self) -> None: - """Should create both catalog PR and issue in one run for catalog-destined test case.""" + async def test_catalog_destined_creates_tracking_issue_after_catalog_pr(self) -> None: + """Should create catalog PR and split-style tracking issue in one run.""" with tempfile.TemporaryDirectory() as tmpdir: test_cases_dir = Path(tmpdir) script_path = "verify_nxos_test.robot" @@ -628,7 +860,7 @@ async def test_catalog_destined_creates_issue_after_catalog_pr(self) -> None: (test_cases_dir / "my_test_cases.yaml").write_text( f"""test_cases: - - title: Test Case 1 + - title: "[NX-OS] Test Case 1" purpose: Test purpose generated_script_path: {script_path} metadata: @@ -668,12 +900,111 @@ async def test_catalog_destined_creates_issue_after_catalog_pr(self) -> None: catalog_repo_url="https://github.com/catalog/repo", ) - # Catalog PR created first, then issue + # Catalog PR created first, then tracking issue assert results["catalog_prs_created"] == 1 assert results["issues_created"] == 1 mock_catalog_adapter.create_pull_request.assert_called_once() mock_project_adapter.create_issue.assert_called_once() + # Verify tracking issue uses split-style template + call_kwargs = mock_project_adapter.create_issue.call_args[1] + assert "Review Catalog PR and Learn Parameters" in call_kwargs["title"] + assert "Catalog PR" in call_kwargs["body"] + assert "Tasks" in call_kwargs["body"] + assert "https://github.com/catalog/repo/pull/20" in call_kwargs["body"] + + @pytest.mark.asyncio + async def test_catalog_destined_uses_tracking_template_from_previous_run(self) -> None: + """Should create tracking issue for catalog-destined test case with pre-existing catalog PR.""" + with tempfile.TemporaryDirectory() as tmpdir: + test_cases_dir = Path(tmpdir) + + # Test case already has catalog_tracking from a previous run but no issue yet + (test_cases_dir / "my_test_cases.yaml").write_text( + """test_cases: + - title: "[IOS-XE] Verify BGP Neighbors" + purpose: Verify BGP neighbors are up + metadata: + catalog: + destined: true + catalog_tracking: + pr_number: 99 + pr_url: https://github.com/catalog/repo/pull/99 + pr_branch: feat/ios-xe/add-verify-iosxe-bgp-neighbors + git_url: https://github.com/catalog/repo + commands: + - command: show bgp summary +""" + ) + + mock_project_adapter = AsyncMock() + mock_issue = MagicMock() + mock_issue.number = 200 + mock_issue.html_url = "https://github.com/org/repo/issues/200" + mock_project_adapter.create_issue.return_value = mock_issue + + with patch( + "github_ops_manager.synchronize.test_requirements.save_test_case_metadata", + return_value=True, + ): + results = await process_test_requirements( + test_cases_dir=test_cases_dir, + base_directory=test_cases_dir, + project_adapter=mock_project_adapter, + project_default_branch="main", + project_repo_url="https://github.com/org/repo", + ) + + # No catalog PR created (already exists), but issue should be created + assert results["catalog_prs_created"] == 0 + assert results["issues_created"] == 1 + + # Verify tracking issue uses split-style template + call_kwargs = mock_project_adapter.create_issue.call_args[1] + assert "Review Catalog PR and Learn Parameters" in call_kwargs["title"] + assert "https://github.com/catalog/repo/pull/99" in call_kwargs["body"] + assert "learn/ios-xe/add-verify-iosxe-bgp-neighbors" in call_kwargs["body"] + + @pytest.mark.asyncio + async def test_non_catalog_uses_collapsed_template(self) -> None: + """Should use collapsed-style template for non-catalog test cases.""" + with tempfile.TemporaryDirectory() as tmpdir: + test_cases_dir = Path(tmpdir) + + (test_cases_dir / "my_test_cases.yaml").write_text( + """test_cases: + - title: Non-Catalog Test + purpose: Test purpose + commands: + - command: show version +""" + ) + + mock_project_adapter = AsyncMock() + mock_issue = MagicMock() + mock_issue.number = 1 + mock_issue.html_url = "https://github.com/org/repo/issues/1" + mock_project_adapter.create_issue.return_value = mock_issue + + with patch( + "github_ops_manager.synchronize.test_requirements.save_test_case_metadata", + return_value=True, + ): + results = await process_test_requirements( + test_cases_dir=test_cases_dir, + base_directory=test_cases_dir, + project_adapter=mock_project_adapter, + project_default_branch="main", + project_repo_url="https://github.com/org/repo", + ) + + assert results["issues_created"] == 1 + + # Verify it uses collapsed style (title is test case title, not tracking format) + call_kwargs = mock_project_adapter.create_issue.call_args[1] + assert call_kwargs["title"] == "Non-Catalog Test" + assert "Review Catalog PR" not in call_kwargs["body"] + @pytest.mark.asyncio async def test_catalog_destined_skips_issue_without_catalog_pr(self) -> None: """Should NOT create issue for catalog-destined test case without generated script.""" From 490ee2b542dd40d8519b1edf4f13433b2d976715 Mon Sep 17 00:00:00 2001 From: Christopher Hart Date: Fri, 6 Feb 2026 09:28:27 -0500 Subject: [PATCH 3/4] fix: apply labels to catalog PRs after creation create_catalog_pr_for_test_case now accepts a labels parameter and calls set_labels_on_issue after PR creation, since the GitHub PR creation API does not support labels inline. The process_test_requirements loop passes test case labels through to catalog PR creation. Co-Authored-By: Claude Opus 4.6 --- .../synchronize/test_requirements.py | 7 ++ .../test_synchronize_test_requirements.py | 78 +++++++++++++++++++ 2 files changed, 85 insertions(+) diff --git a/github_ops_manager/synchronize/test_requirements.py b/github_ops_manager/synchronize/test_requirements.py index 42f21f1..5b6aef7 100644 --- a/github_ops_manager/synchronize/test_requirements.py +++ b/github_ops_manager/synchronize/test_requirements.py @@ -366,6 +366,7 @@ async def create_catalog_pr_for_test_case( base_directory: Path, default_branch: str, catalog_repo_url: str, + labels: list[str] | None = None, ) -> dict[str, Any] | None: """Create a catalog PR for a test case and update metadata. @@ -375,6 +376,7 @@ async def create_catalog_pr_for_test_case( base_directory: Base directory for resolving file paths default_branch: Default branch to base PR on catalog_repo_url: Full URL to catalog repository + labels: Optional list of labels to apply to the PR Returns: Created PR data or None on error @@ -454,6 +456,10 @@ async def create_catalog_pr_for_test_case( pr_url=new_pr.html_url, ) + # Apply labels to PR (GitHub treats PRs as issues for labels) + if labels: + await github_adapter.set_labels_on_issue(new_pr.number, labels) + # Update test case with catalog PR metadata update_test_case_with_pr_metadata(test_case, new_pr, catalog_repo_url) @@ -621,6 +627,7 @@ async def process_test_requirements( base_directory, catalog_default_branch, catalog_repo_url, + labels=test_case.get("labels", issue_labels), ) if catalog_pr_result: diff --git a/tests/unit/test_synchronize_test_requirements.py b/tests/unit/test_synchronize_test_requirements.py index 1e66252..4b5a8d4 100644 --- a/tests/unit/test_synchronize_test_requirements.py +++ b/tests/unit/test_synchronize_test_requirements.py @@ -407,6 +407,84 @@ async def test_extracts_os_from_filename_fallback(self) -> None: assert result is not None assert result["catalog_path"] == "catalog/IOS-XE/verify_ios_xe_interfaces.robot" + @pytest.mark.asyncio + async def test_applies_labels_to_catalog_pr(self) -> None: + """Should apply labels to the catalog PR after creation.""" + with tempfile.TemporaryDirectory() as tmpdir: + base_dir = Path(tmpdir) + script_path = "verify_nxos_interfaces.robot" + robot_content = """*** Settings *** +Test Tags os:nxos category:foundations + +*** Test Cases *** +Test + Log Hello +""" + (base_dir / script_path).write_text(robot_content) + + mock_adapter = AsyncMock() + mock_adapter.branch_exists.return_value = False + mock_pr = MagicMock() + mock_pr.number = 101 + mock_pr.html_url = "https://github.com/catalog/repo/pull/101" + mock_pr.head.ref = "feat/nxos/add-verify-nxos-interfaces" + mock_adapter.create_pull_request.return_value = mock_pr + + test_case: dict[str, Any] = { + "title": "Test Case 1", + "generated_script_path": script_path, + } + + result = await create_catalog_pr_for_test_case( + test_case, + mock_adapter, + base_dir, + "main", + "https://github.com/catalog/repo", + labels=["quicksilver", "GenAI"], + ) + + assert result is not None + mock_adapter.set_labels_on_issue.assert_called_once_with(101, ["quicksilver", "GenAI"]) + + @pytest.mark.asyncio + async def test_skips_labels_when_none(self) -> None: + """Should not call set_labels_on_issue when no labels provided.""" + with tempfile.TemporaryDirectory() as tmpdir: + base_dir = Path(tmpdir) + script_path = "verify_nxos_interfaces.robot" + robot_content = """*** Settings *** +Test Tags os:nxos + +*** Test Cases *** +Test + Log Hello +""" + (base_dir / script_path).write_text(robot_content) + + mock_adapter = AsyncMock() + mock_adapter.branch_exists.return_value = False + mock_pr = MagicMock() + mock_pr.number = 101 + mock_pr.html_url = "https://github.com/catalog/repo/pull/101" + mock_pr.head.ref = "feat/nxos/add-verify-nxos-interfaces" + mock_adapter.create_pull_request.return_value = mock_pr + + test_case: dict[str, Any] = { + "title": "Test Case 1", + "generated_script_path": script_path, + } + + await create_catalog_pr_for_test_case( + test_case, + mock_adapter, + base_dir, + "main", + "https://github.com/catalog/repo", + ) + + mock_adapter.set_labels_on_issue.assert_not_called() + @pytest.mark.asyncio async def test_returns_none_when_os_not_detected(self) -> None: """Should return None if OS cannot be extracted.""" From 0db920df5c6efabbfe608a392af2fc6f44874d3d Mon Sep 17 00:00:00 2001 From: Christopher Hart Date: Fri, 6 Feb 2026 09:40:42 -0500 Subject: [PATCH 4/4] fix: add quicksilver label and remove script-already-created from catalog PRs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Catalog PRs now get the "quicksilver" label added and the "script-already-created" label removed before label application. Issue labels are unchanged — they continue to use whatever is defined on the test case. Co-Authored-By: Claude Opus 4.6 --- .../synchronize/test_requirements.py | 10 ++- .../test_synchronize_test_requirements.py | 75 +++++++++++++++++++ 2 files changed, 84 insertions(+), 1 deletion(-) diff --git a/github_ops_manager/synchronize/test_requirements.py b/github_ops_manager/synchronize/test_requirements.py index 5b6aef7..155e401 100644 --- a/github_ops_manager/synchronize/test_requirements.py +++ b/github_ops_manager/synchronize/test_requirements.py @@ -621,13 +621,21 @@ async def process_test_requirements( results["errors"].append(f"Catalog PR needed for {title} but catalog not configured") continue + # Build catalog PR labels: start from test case / default labels, + # remove issue-specific labels, and add "quicksilver" + catalog_pr_labels = list(test_case.get("labels", issue_labels) or []) + if "script-already-created" in catalog_pr_labels: + catalog_pr_labels.remove("script-already-created") + if "quicksilver" not in catalog_pr_labels: + catalog_pr_labels.append("quicksilver") + catalog_pr_result = await create_catalog_pr_for_test_case( test_case, catalog_adapter, base_directory, catalog_default_branch, catalog_repo_url, - labels=test_case.get("labels", issue_labels), + labels=catalog_pr_labels, ) if catalog_pr_result: diff --git a/tests/unit/test_synchronize_test_requirements.py b/tests/unit/test_synchronize_test_requirements.py index 4b5a8d4..16cb908 100644 --- a/tests/unit/test_synchronize_test_requirements.py +++ b/tests/unit/test_synchronize_test_requirements.py @@ -941,6 +941,9 @@ async def test_catalog_destined_creates_tracking_issue_after_catalog_pr(self) -> - title: "[NX-OS] Test Case 1" purpose: Test purpose generated_script_path: {script_path} + labels: + - GenAI + - script-already-created metadata: catalog: destined: true @@ -991,6 +994,78 @@ async def test_catalog_destined_creates_tracking_issue_after_catalog_pr(self) -> assert "Tasks" in call_kwargs["body"] assert "https://github.com/catalog/repo/pull/20" in call_kwargs["body"] + # Verify catalog PR gets "quicksilver" label, not "script-already-created" + catalog_label_call = mock_catalog_adapter.set_labels_on_issue.call_args[0] + catalog_labels = catalog_label_call[1] + assert "quicksilver" in catalog_labels + assert "GenAI" in catalog_labels + assert "script-already-created" not in catalog_labels + + @pytest.mark.asyncio + async def test_catalog_pr_labels_transformed_from_issue_labels(self) -> None: + """Should add 'quicksilver' and remove 'script-already-created' from catalog PR labels.""" + with tempfile.TemporaryDirectory() as tmpdir: + test_cases_dir = Path(tmpdir) + script_path = "verify_nxos_test.robot" + robot_content = """*** Settings *** +Test Tags os:nxos + +*** Test Cases *** +Test + Log Hello +""" + (test_cases_dir / script_path).write_text(robot_content) + + (test_cases_dir / "my_test_cases.yaml").write_text( + f"""test_cases: + - title: Test Case 1 + purpose: Test purpose + generated_script_path: {script_path} + labels: + - GenAI + - script-already-created + metadata: + catalog: + destined: true + project_tracking: + issue_number: 1 + issue_url: https://existing-url + commands: + - command: show version +""" + ) + + mock_project_adapter = AsyncMock() + mock_catalog_adapter = AsyncMock() + mock_catalog_adapter.branch_exists.return_value = False + mock_pr = MagicMock() + mock_pr.number = 30 + mock_pr.html_url = "https://github.com/catalog/repo/pull/30" + mock_pr.head.ref = "feat/nxos/add-test" + mock_catalog_adapter.create_pull_request.return_value = mock_pr + + with patch( + "github_ops_manager.synchronize.test_requirements.save_test_case_metadata", + return_value=True, + ): + results = await process_test_requirements( + test_cases_dir=test_cases_dir, + base_directory=test_cases_dir, + project_adapter=mock_project_adapter, + project_default_branch="main", + project_repo_url="https://github.com/org/repo", + catalog_adapter=mock_catalog_adapter, + catalog_default_branch="main", + catalog_repo_url="https://github.com/catalog/repo", + ) + + assert results["catalog_prs_created"] == 1 + + # Verify label transformation: quicksilver added, script-already-created removed + catalog_label_call = mock_catalog_adapter.set_labels_on_issue.call_args[0] + catalog_labels = catalog_label_call[1] + assert catalog_labels == ["GenAI", "quicksilver"] + @pytest.mark.asyncio async def test_catalog_destined_uses_tracking_template_from_previous_run(self) -> None: """Should create tracking issue for catalog-destined test case with pre-existing catalog PR."""