Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions github_ops_manager/configuration/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -256,6 +256,14 @@ def process_test_requirements_cli(
"If provided, migrates existing issue/PR metadata to test_cases.yaml before processing.",
),
] = None,
cxtm_file: Annotated[
Path | None,
Option(
envvar="CXTM_FILE",
help="Path to cxtm.yaml file. If provided, enriches issue bodies with test case "
"definitions from cxtm.yaml (title, robot_file, git_commit_sha, etc.).",
),
] = None,
) -> None:
"""Process test requirements directly from test_cases.yaml files.

Expand Down Expand Up @@ -404,6 +412,7 @@ async def run_processing() -> dict:
catalog_repo_url=catalog_repo_url,
issue_template_path=issue_template,
issue_labels=parsed_labels,
cxtm_file_path=cxtm_file,
)

results = asyncio.run(run_processing())
Expand Down
38 changes: 37 additions & 1 deletion github_ops_manager/schemas/tac.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Schemas/models for Testing as Code constructs."""

from pydantic import BaseModel
from pydantic import BaseModel, Field


class TestingAsCodeCommand(BaseModel):
Expand Down Expand Up @@ -33,3 +33,39 @@ class TestingAsCodeTestCaseDefinitions(BaseModel):
"""Pydantic model for a list of Testing as Code test case definitions."""

test_cases: list[TestingAsCodeTestCaseDefinition]


# =============================================================================
# CXTM Configuration Models (cxtm.yaml)
# =============================================================================
# These models represent the test case definitions stored in cxtm.yaml,
# which is the authoritative source for test case metadata used by tac-tools.


class CXTMTestCase(BaseModel):
"""A single test case within a test case group in cxtm.yaml."""

identifier: str = Field(description="Unique identifier within the group (e.g., '0')")
title: str = Field(description="Human-readable test case title")
robot_file: str | None = Field(default=None, description="Path to the Robot Framework test file")
git_url: str | None = Field(default=None, description="URL of the git repository containing the test")
git_commit_sha: str | None = Field(default=None, description="Git commit SHA or branch name")
parameters_file: str | None = Field(default=None, description="Path to parameters JSON file")
parameters: dict | None = Field(default=None, description="Inline test parameters")


class CXTMTestCaseGroup(BaseModel):
"""A test case group in cxtm.yaml containing one or more test cases."""

name: str = Field(description="Name of the test case group")
test_cases: list[CXTMTestCase] = Field(default_factory=list, description="List of test cases in this group")


class CXTMConfiguration(BaseModel):
"""Root model for cxtm.yaml configuration file."""

cxta_version: str | None = Field(default=None, description="CXTA version")
cxtm_project_id: str | None = Field(default=None, description="CXTM project ID")
git_branch: str | None = Field(default=None, description="Default git branch")
git_url: str | None = Field(default=None, description="Project repository URL")
test_case_groups: list[CXTMTestCaseGroup] = Field(default_factory=list, description="List of test case groups")
43 changes: 42 additions & 1 deletion github_ops_manager/synchronize/test_requirements.py
Original file line number Diff line number Diff line change
Expand Up @@ -503,13 +503,15 @@ def render_issue_body_for_test_case(
test_case: dict[str, Any],
template: jinja2.Template,
max_body_length: int | None = None,
cxtm_config: Any | None = None,
) -> str:
"""Render issue body for a test case using the template.

Args:
test_case: Test case dictionary with all fields
template: Jinja2 template for issue body
max_body_length: Optional max length for issue body (truncates outputs if needed)
cxtm_config: Optional CXTMConfiguration for enriching with test case definitions

Returns:
Rendered issue body string
Expand All @@ -528,6 +530,23 @@ def render_issue_body_for_test_case(
"commands": commands_as_dicts,
}

# Look up test case definition in cxtm.yaml and add to render context
if cxtm_config:
from github_ops_manager.utils.cxtm import find_test_case_by_title, format_test_case_yaml_block

title = test_case.get("title", "")
group, cxtm_test_case = find_test_case_by_title(cxtm_config, title)
if group and cxtm_test_case:
render_context["test_case_definition"] = format_test_case_yaml_block(group, cxtm_test_case)
logger.debug(
"Found cxtm.yaml test case definition",
title=title,
group_name=group.name,
test_case_id=cxtm_test_case.identifier,
)
else:
logger.debug("No cxtm.yaml test case definition found", title=title)

# Apply truncation to command outputs if max_body_length is specified
if max_body_length is not None:
render_context = truncate_data_dict_outputs(render_context, max_body_length)
Expand All @@ -551,6 +570,7 @@ async def process_test_requirements(
issue_template_path: Path | None = None,
issue_labels: list[str] | None = None,
max_body_length: int = DEFAULT_MAX_ISSUE_BODY_LENGTH,
cxtm_file_path: Path | None = None,
) -> dict[str, Any]:
"""Process all test requirements: create issues and PRs as needed.

Expand All @@ -569,6 +589,7 @@ async def process_test_requirements(
issue_template_path: Optional path to Jinja2 template for issue bodies
issue_labels: Optional list of labels to apply to issues
max_body_length: Maximum issue body length (truncates outputs if exceeded)
cxtm_file_path: Optional path to cxtm.yaml for enriching issues with test case definitions

Returns:
Summary dict with counts and results
Expand All @@ -579,6 +600,26 @@ async def process_test_requirements(
base_directory=str(base_directory),
)

# Load cxtm.yaml configuration if provided
cxtm_config = None
if cxtm_file_path:
from github_ops_manager.utils.cxtm import load_cxtm_configuration

try:
cxtm_config = load_cxtm_configuration(cxtm_file_path)
logger.info(
"Loaded cxtm.yaml configuration",
cxtm_file=str(cxtm_file_path),
test_case_groups=len(cxtm_config.test_case_groups),
)
except Exception as e:
logger.warning(
"Failed to load cxtm.yaml, proceeding without test case definitions",
cxtm_file=str(cxtm_file_path),
error=str(e),
)
cxtm_config = None

# Load issue body template if provided
template = None
if issue_template_path:
Expand Down Expand Up @@ -666,7 +707,7 @@ async def process_test_requirements(
# 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)
issue_body = render_issue_body_for_test_case(test_case, template, max_body_length=max_body_length, cxtm_config=cxtm_config)
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}")
Expand Down
10 changes: 10 additions & 0 deletions github_ops_manager/templates/tac_issues_body.j2
Original file line number Diff line number Diff line change
@@ -1,3 +1,13 @@
{% if test_case_definition is defined and test_case_definition %}
## Test Case Definition

```yaml
{{ test_case_definition }}
```

{% endif %}
## Test Requirement

{% if purpose is defined %}
**Purpose**: {{ purpose }}

Expand Down
166 changes: 166 additions & 0 deletions github_ops_manager/utils/cxtm.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,166 @@
"""Utility functions for parsing and working with cxtm.yaml files.

The cxtm.yaml file is the authoritative source for test case definitions
used by tac-tools. This module provides functions to extract test case
metadata that can be embedded into GitHub issues for traceability.
"""

import logging
import re
from pathlib import Path

from github_ops_manager.schemas.tac import (
CXTMConfiguration,
CXTMTestCase,
CXTMTestCaseGroup,
)
from github_ops_manager.utils.yaml import load_yaml_file

logger = logging.getLogger(__name__)


def load_cxtm_configuration(cxtm_file_path: Path) -> CXTMConfiguration:
"""Load and parse a cxtm.yaml configuration file.

Args:
cxtm_file_path: Path to the cxtm.yaml file.

Returns:
CXTMConfiguration object with parsed data.

Raises:
FileNotFoundError: If the file doesn't exist.
ValidationError: If the file has invalid structure.
"""
if not cxtm_file_path.exists():
raise FileNotFoundError(f"cxtm.yaml not found: {cxtm_file_path.absolute()}")

yaml_content = load_yaml_file(cxtm_file_path)
return CXTMConfiguration.model_validate(yaml_content)


def find_test_case_by_title(
cxtm_config: CXTMConfiguration,
title: str,
fuzzy_match: bool = True,
) -> tuple[CXTMTestCaseGroup | None, CXTMTestCase | None]:
"""Find a test case in cxtm.yaml by its title.

Searches through all test case groups to find a test case matching
the given title. Supports both exact and fuzzy matching.

Args:
cxtm_config: Parsed cxtm.yaml configuration.
title: Title to search for (e.g., "[NX-OS] Verify Module Serial Number").
fuzzy_match: If True, performs case-insensitive partial matching.

Returns:
Tuple of (test_case_group, test_case) if found, (None, None) otherwise.
"""
# Normalize title for comparison
normalized_title = title.strip().lower() if fuzzy_match else title.strip()

for group in cxtm_config.test_case_groups:
# Check group name
group_name = group.name.strip().lower() if fuzzy_match else group.name.strip()

if fuzzy_match:
# Check if title matches group name or is contained within it
if normalized_title in group_name or group_name in normalized_title:
if group.test_cases:
return group, group.test_cases[0]
else:
if group_name == normalized_title:
if group.test_cases:
return group, group.test_cases[0]

# Also check individual test case titles within the group
for test_case in group.test_cases:
tc_title = test_case.title.strip().lower() if fuzzy_match else test_case.title.strip()
if fuzzy_match:
if normalized_title in tc_title or tc_title in normalized_title:
return group, test_case
else:
if tc_title == normalized_title:
return group, test_case

return None, None


def find_test_case_by_robot_file(
cxtm_config: CXTMConfiguration,
robot_file_path: str,
) -> tuple[CXTMTestCaseGroup | None, CXTMTestCase | None]:
"""Find a test case in cxtm.yaml by its robot file path.

Args:
cxtm_config: Parsed cxtm.yaml configuration.
robot_file_path: Path to the robot file (can be partial path).

Returns:
Tuple of (test_case_group, test_case) if found, (None, None) otherwise.
"""
# Normalize path for comparison
normalized_path = robot_file_path.strip().lower()

for group in cxtm_config.test_case_groups:
for test_case in group.test_cases:
if test_case.robot_file:
tc_robot_file = test_case.robot_file.strip().lower()
# Check for exact match or if one contains the other
if normalized_path in tc_robot_file or tc_robot_file in normalized_path:
return group, test_case

return None, None


def extract_test_name_from_title(title: str) -> str:
"""Extract the test name from a title in format '[OS] Test Name'.

Args:
title: Full title string (e.g., "[NX-OS] Verify Module Serial Number").

Returns:
The test name portion (e.g., "Verify Module Serial Number").
"""
pattern = re.compile(r"^\[(?P<os>[^\]]+)\]\s+(?P<test_name>.+)$")
match = pattern.match(title.strip())
if match:
return match.group("test_name")
return title.strip()


def format_test_case_yaml_block(
group: CXTMTestCaseGroup,
test_case: CXTMTestCase,
) -> str:
"""Format a test case definition as a YAML block for embedding in issues.

Creates a structured YAML representation of the test case that can be
parsed by the webhook receiver to extract test case metadata.

Args:
group: The test case group containing the test case.
test_case: The test case to format.

Returns:
Formatted YAML string ready for embedding in an issue body.
"""
lines = [
"test_case_group:",
f' name: "{group.name}"',
"test_case:",
f' identifier: "{test_case.identifier}"',
f' title: "{test_case.title}"',
]

if test_case.robot_file:
lines.append(f' robot_file: "{test_case.robot_file}"')

if test_case.git_url:
lines.append(f' git_url: "{test_case.git_url}"')

if test_case.git_commit_sha:
lines.append(f' git_commit_sha: "{test_case.git_commit_sha}"')

return "\n".join(lines)