PyTorch HUD MCP Server
by izaitsevfb
- test
#!/usr/bin/env python3
"""
Test script that injects failures into the sample data to verify that failure detection works correctly.
"""
import json
import unittest
from unittest.mock import patch, MagicMock, AsyncMock
import copy
from pytorch_hud.tools.hud_data import get_recent_commits_with_jobs
class TestFailureInjection(unittest.IsolatedAsyncioTestCase):
"""Tests that inject failures into sample data to validate detection logic."""
def setUp(self):
"""Load sample HUD data from file and inject failures."""
# Path to the sample data file
sample_file_path = "test/fixtures/hud_data_response_sample_per_page_50.json"
try:
with open(sample_file_path, 'r') as f:
self.original_data = json.load(f)
except Exception as e:
self.fail(f"Failed to load sample data: {e}")
# Create a copy of the data to modify
self.data_with_failures = copy.deepcopy(self.original_data)
# Inject failures into the data
# 1. Inject an explicit failure job
explicit_failure = {
"sha": "e6800bda7fabaf1de7c0586c9851c2326e142993",
"id": 99999001,
"conclusion": "failure",
"status": "completed",
"htmlUrl": "https://github.com/pytorch/pytorch/actions/runs/13709183265/job/99999001",
"logUrl": "https://ossci-raw-job-status.s3.amazonaws.com/log/99999001",
"durationS": 120,
"repo": "pytorch/pytorch",
"failureLines": ["ERROR: Test failed with exit code 1", "Error: Compilation failed"],
"failureLineNumbers": [100, 200],
"failureCaptures": ["Test failed", "Compilation failed"]
}
# 2. Inject a hidden failure job (success conclusion but with failure lines)
hidden_failure = {
"sha": "e6800bda7fabaf1de7c0586c9851c2326e142993",
"id": 99999002,
"conclusion": "success",
"status": "completed",
"htmlUrl": "https://github.com/pytorch/pytorch/actions/runs/13709183265/job/99999002",
"logUrl": "https://ossci-raw-job-status.s3.amazonaws.com/log/99999002",
"durationS": 150,
"repo": "pytorch/pytorch",
"failureLines": ["WARNING: Test output does not match expected results", "ERROR: Floating point exception"],
"failureLineNumbers": [300, 400],
"failureCaptures": ["Test mismatch", "FP exception"]
}
# 3. Inject a test failure job for testing get_test_summary
test_failure = {
"sha": "e6800bda7fabaf1de7c0586c9851c2326e142993",
"id": 99999003,
"conclusion": "failure",
"status": "completed",
"htmlUrl": "https://github.com/pytorch/pytorch/actions/runs/13709183265/test_workflow/job/99999003",
"logUrl": "https://ossci-raw-job-status.s3.amazonaws.com/log/99999003",
"durationS": 180,
"repo": "pytorch/pytorch",
"failureLines": [
"FAIL: test_function1 (test_module.TestClass)",
"ERROR: test_function2 (test_module.TestClass)",
"AssertionError: Values not equal: 1 != 2"
],
"failureLineNumbers": [500, 600, 700],
"failureCaptures": ["Test failure", "Test error", "Assertion"]
}
# Add jobs to the first entry in shaGrid
if len(self.data_with_failures["shaGrid"]) > 0 and "jobs" in self.data_with_failures["shaGrid"][0]:
# Add the failure jobs to the beginning to make them easier to find
self.data_with_failures["shaGrid"][0]["jobs"].insert(0, explicit_failure)
self.data_with_failures["shaGrid"][0]["jobs"].insert(1, hidden_failure)
self.data_with_failures["shaGrid"][0]["jobs"].insert(2, test_failure)
# Log the number of jobs now in the test data
success_jobs = [job for job in self.data_with_failures["shaGrid"][0]["jobs"]
if job.get("conclusion") == "success" and not
(job.get("failureLines", []) and len(job.get("failureLines", [])) > 0)]
explicit_failure_jobs = [job for job in self.data_with_failures["shaGrid"][0]["jobs"]
if job.get("conclusion") == "failure"]
hidden_failure_jobs = [job for job in self.data_with_failures["shaGrid"][0]["jobs"]
if job.get("conclusion") == "success" and
job.get("failureLines", []) and len(job.get("failureLines", [])) > 0]
print("Test data contains:")
print(f"- {len(success_jobs)} true success jobs")
print(f"- {len(explicit_failure_jobs)} explicit failure jobs")
print(f"- {len(hidden_failure_jobs)} hidden failure jobs")
else:
self.fail("Could not find jobs array in sample data")
@patch('pytorch_hud.tools.hud_data.api.get_hud_data')
async def test_failure_detection(self, mock_get_hud_data):
"""Test that get_recent_commits_with_jobs correctly detects failures."""
# Mock the API to return our modified data
mock_get_hud_data.return_value = self.data_with_failures
# Create a mock context with async methods
mock_ctx = MagicMock()
mock_ctx.info = AsyncMock()
# Call get_recent_commits_with_jobs with include_failures=True
result = await get_recent_commits_with_jobs(
"pytorch",
"pytorch",
"main",
include_failures=True,
ctx=mock_ctx
)
# Verify we have commits in the result
self.assertTrue(result["commits"], "No commits returned in result")
self.assertGreaterEqual(len(result["commits"]), 1, "Expected at least 1 commit")
# Get the first commit
commit = result["commits"][0]
# Verify job counts - should include our injected failures
self.assertIn("job_counts", commit, "No job counts found in commit data")
self.assertIn("failure", commit["job_counts"], "No failure count found")
self.assertGreaterEqual(commit["job_counts"]["failure"], 2,
f"Expected at least 2 failures, got {commit['job_counts']['failure']}")
# Verify the commit status is red due to failures
self.assertEqual(commit["status"], "red", "Commit status should be 'red' due to failures")
# Verify jobs are included and contain failures
self.assertIn("jobs", commit, "No jobs array in commit data")
self.assertGreaterEqual(len(commit["jobs"]), 2, f"Expected at least 2 jobs, got {len(commit['jobs'])}")
# Extract job IDs
job_ids = [job.get("id") for job in commit["jobs"]]
# Check for our injected failure job IDs
# Job IDs might be returned as strings, so convert to strings for comparison
job_id_strings = [str(job_id) for job_id in job_ids]
self.assertTrue(
"99999001" in job_id_strings or "99999003" in job_id_strings,
"Neither of our explicit failure jobs (99999001, 99999003) found in result"
)
# The hidden failure job should NOT be included because it has conclusion=success
self.assertNotIn("99999002", job_id_strings,
"Hidden failure job (99999002) should not be in failures list")
# Print results for debugging
print("Failure Detection Results:")
print(f"- Commit status: {commit['status']}")
print(f"- Job counts: {commit['job_counts']}")
print(f"- Number of jobs returned: {len(commit['jobs'])}")
print(f"- Job IDs: {job_ids}")
@patch('pytorch_hud.tools.hud_data.api.get_hud_data')
async def test_job_filtering(self, mock_get_hud_data):
"""Test that job filtering options work correctly."""
# Mock the API to return our modified data
mock_get_hud_data.return_value = self.data_with_failures
# Create a mock context with async methods
mock_ctx = MagicMock()
mock_ctx.info = AsyncMock()
# Call with job_name_filter_regex to filter for test jobs
result = await get_recent_commits_with_jobs(
"pytorch",
"pytorch",
"main",
include_failures=True,
job_name_filter_regex="test_workflow",
ctx=mock_ctx
)
# Verify that filtering worked
self.assertTrue(result["commits"], "No commits returned in result")
commit = result["commits"][0]
# Should have filtered to only include test jobs
if "jobs" in commit and commit["jobs"]:
# For any jobs that are returned, verify they match the filter
for job in commit["jobs"]:
# Check job name or URL for the test_workflow pattern
job_name = job.get("name", "")
html_url = job.get("htmlUrl", "")
# If not directly in name, check the URL
if "test_workflow" not in job_name and html_url:
self.assertIn("test_workflow", html_url,
f"Job {job.get('id')} doesn't match the filter pattern")
# Print results for debugging
print("Job Filtering Results:")
print(f"- Commit status: {commit['status']}")
if "jobs" in commit:
print(f"- Number of filtered jobs: {len(commit['jobs'])}")
print(f"- Job IDs: {[job.get('id') for job in commit['jobs']]}")
@patch('pytorch_hud.tools.hud_data.api.get_hud_data')
async def test_failure_line_filtering(self, mock_get_hud_data):
"""Test filtering by failure line content."""
# Mock the API to return our modified data
mock_get_hud_data.return_value = self.data_with_failures
# Create a mock context with async methods
mock_ctx = MagicMock()
mock_ctx.info = AsyncMock()
# Call with failure line filter to find only compilation errors
result = await get_recent_commits_with_jobs(
"pytorch",
"pytorch",
"main",
include_failures=True,
failure_line_filter_regex="Compilation failed",
ctx=mock_ctx
)
# Verify that filtering worked
self.assertTrue(result["commits"], "No commits returned in result")
commit = result["commits"][0]
# If jobs are found, they should have the specified failure line
if "jobs" in commit and commit["jobs"]:
for job in commit["jobs"]:
failure_lines = job.get("failureLines", [])
if failure_lines:
compilation_failure_found = any("Compilation failed" in line for line in failure_lines)
self.assertTrue(compilation_failure_found,
f"Job {job.get('id')} doesn't have the expected failure line")
# Print results for debugging
print("Failure Line Filtering Results:")
if "jobs" in commit:
print(f"- Number of jobs with compilation failures: {len(commit.get('jobs', []))}")
for job in commit.get("jobs", []):
print(f"- Job {job.get('id')} failure lines: {job.get('failureLines', [])}")
if __name__ == "__main__":
unittest.main()