xiaomai
c73767073e
Catch Gemini API errors (e.g., HTTP 429) and summarize upstream messages Generate fallback markdown for failed files instead of aborting Append analysis failures to pipeline warnings
179 lines
8.4 KiB
Python
179 lines
8.4 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
import sys
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
from uipath_explainator.pipeline import ProjectPipeline
|
|
from uipath_explainator.scanner import crawl_dependencies, extract_dependencies, strip_comment_out_blocks
|
|
|
|
|
|
MAIN_XAML = """<?xml version="1.0" encoding="utf-8"?>
|
|
<Activity xmlns="http://schemas.microsoft.com/netfx/2009/xaml/activities"
|
|
xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
|
|
xmlns:ui="http://schemas.uipath.com/workflow/activities">
|
|
<Sequence>
|
|
<ui:InvokeWorkflowFile WorkflowFileName="Flows/Active.xaml" />
|
|
<ui:CommentOut>
|
|
<Sequence>
|
|
<ui:InvokeWorkflowFile WorkflowFileName="Flows/Old.xaml" />
|
|
</Sequence>
|
|
</ui:CommentOut>
|
|
</Sequence>
|
|
</Activity>
|
|
"""
|
|
|
|
ACTIVE_XAML = """<?xml version="1.0" encoding="utf-8"?>
|
|
<Activity xmlns="http://schemas.microsoft.com/netfx/2009/xaml/activities"
|
|
xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
|
|
xmlns:ui="http://schemas.uipath.com/workflow/activities"
|
|
xmlns:sap2010="http://schemas.microsoft.com/netfx/2010/xaml/activities/presentation">
|
|
<Sequence>
|
|
<ui:InvokeVBA>
|
|
<ui:InvokeVBA.CodeFilePath>
|
|
<InArgument x:TypeArguments="x:String">["Scripts/Keep.bas"]</InArgument>
|
|
</ui:InvokeVBA.CodeFilePath>
|
|
</ui:InvokeVBA>
|
|
</Sequence>
|
|
</Activity>
|
|
"""
|
|
|
|
OLD_XAML = """<?xml version="1.0" encoding="utf-8"?>
|
|
<Activity xmlns="http://schemas.microsoft.com/netfx/2009/xaml/activities"
|
|
xmlns:ui="http://schemas.uipath.com/workflow/activities">
|
|
<Sequence>
|
|
<ui:InvokeVBA CodeFilePath="Scripts/Drop.bas" />
|
|
</Sequence>
|
|
</Activity>
|
|
"""
|
|
|
|
|
|
class StubAnalyzer:
|
|
def analyze(self, relative_path: Path, content: str) -> str:
|
|
return f"# {relative_path.as_posix()}\n\n{len(content)}"
|
|
|
|
|
|
class FlakyAnalyzer:
|
|
def analyze(self, relative_path: Path, content: str) -> str:
|
|
if relative_path.name == "Active.xaml":
|
|
raise RuntimeError("HTTP 429 quota exhausted")
|
|
return f"# {relative_path.as_posix()}\n\n{len(content)}"
|
|
|
|
|
|
class PipelineTests(unittest.TestCase):
|
|
def test_strip_comment_out_blocks_removes_nested_blocks(self) -> None:
|
|
source = "<root><ui:CommentOut><x/><ui:CommentOut><y/></ui:CommentOut></ui:CommentOut><z/></root>"
|
|
cleaned = strip_comment_out_blocks(source)
|
|
self.assertEqual(cleaned, "<root><z/></root>")
|
|
|
|
def test_extract_dependencies_reads_property_element_paths(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
flows = root / "Flows"
|
|
flows.mkdir()
|
|
(root / "Main.xaml").write_text(ACTIVE_XAML, encoding="utf-8")
|
|
(root / "Scripts").mkdir()
|
|
(root / "Scripts" / "Keep.bas").write_text("Sub Keep()", encoding="utf-8")
|
|
|
|
dependencies, warnings = extract_dependencies(root, root / "Main.xaml")
|
|
|
|
self.assertEqual(len(warnings), 0)
|
|
self.assertEqual(len(dependencies), 1)
|
|
self.assertEqual(dependencies[0].target, root / "Scripts" / "Keep.bas")
|
|
|
|
def test_pipeline_prunes_files_only_reachable_before_comment_cleanup(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
project_root = tmp_path / "project"
|
|
output_root = tmp_path / "workspace"
|
|
code_root = output_root / "code"
|
|
docs_root = output_root / "docs"
|
|
(project_root / "Flows").mkdir(parents=True)
|
|
(project_root / "Flows" / "Active.xaml").write_text(ACTIVE_XAML, encoding="utf-8")
|
|
(project_root / "Flows" / "Old.xaml").write_text(OLD_XAML, encoding="utf-8")
|
|
(project_root / "Scripts").mkdir()
|
|
(project_root / "Scripts" / "Keep.bas").write_text("Sub Keep()\nEnd Sub", encoding="utf-8")
|
|
(project_root / "Scripts" / "Drop.bas").write_text("Sub Drop()\nEnd Sub", encoding="utf-8")
|
|
(project_root / "main.xaml").write_text(MAIN_XAML, encoding="utf-8")
|
|
|
|
initial_scan = crawl_dependencies(project_root, project_root / "main.xaml")
|
|
initial_files = {path.relative_to(project_root).as_posix() for path in initial_scan.files}
|
|
self.assertIn("Flows/Old.xaml", initial_files)
|
|
self.assertIn("Scripts/Drop.bas", initial_files)
|
|
|
|
report = ProjectPipeline(project_root, output_root, "main.xaml", force=True).run(StubAnalyzer())
|
|
|
|
final_files = {path.as_posix() for path in report.final_files}
|
|
self.assertIn("main.xaml", final_files)
|
|
self.assertIn("Flows/Active.xaml", final_files)
|
|
self.assertIn("Scripts/Keep.bas", final_files)
|
|
self.assertNotIn("Flows/Old.xaml", final_files)
|
|
self.assertNotIn("Scripts/Drop.bas", final_files)
|
|
self.assertEqual(report.code_root, code_root.resolve())
|
|
self.assertEqual(report.docs_root, docs_root.resolve())
|
|
self.assertFalse((code_root / "Flows" / "Old.xaml").exists())
|
|
self.assertFalse((code_root / "Scripts" / "Drop.bas").exists())
|
|
self.assertTrue((code_root / "Flows" / "Active.xaml").exists())
|
|
self.assertTrue((docs_root / "Flows" / "Active.xaml.analysis.md").exists())
|
|
self.assertTrue((docs_root / "manifest.json").exists())
|
|
|
|
overview = (docs_root / "OVERVIEW.md").read_text(encoding="utf-8")
|
|
self.assertIn("## Processing Logic", overview)
|
|
self.assertIn("Initial Scan", overview)
|
|
self.assertIn("## How To Read This Output", overview)
|
|
self.assertIn("## Cleaned XAML Files", overview)
|
|
self.assertIn("Code Root", overview)
|
|
self.assertIn("Docs Root", overview)
|
|
|
|
def test_pipeline_emits_stage_logs(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
project_root = tmp_path / "project"
|
|
output_root = tmp_path / "workspace"
|
|
(project_root / "Flows").mkdir(parents=True)
|
|
(project_root / "Flows" / "Active.xaml").write_text(ACTIVE_XAML, encoding="utf-8")
|
|
(project_root / "Flows" / "Old.xaml").write_text(OLD_XAML, encoding="utf-8")
|
|
(project_root / "Scripts").mkdir()
|
|
(project_root / "Scripts" / "Keep.bas").write_text("Sub Keep()\nEnd Sub", encoding="utf-8")
|
|
(project_root / "Scripts" / "Drop.bas").write_text("Sub Drop()\nEnd Sub", encoding="utf-8")
|
|
(project_root / "main.xaml").write_text(MAIN_XAML, encoding="utf-8")
|
|
|
|
with self.assertLogs("uipath_explainator", level="INFO") as captured:
|
|
ProjectPipeline(project_root, output_root, "main.xaml", force=True).run(StubAnalyzer())
|
|
|
|
combined = "\n".join(captured.output)
|
|
self.assertIn("Starting pipeline:", combined)
|
|
self.assertIn("Initial scan complete:", combined)
|
|
self.assertIn("Copied 5 files and cleaned 1 XAML files", combined)
|
|
self.assertIn("Final scan complete:", combined)
|
|
self.assertIn("Pipeline completed in", combined)
|
|
|
|
def test_pipeline_keeps_running_when_single_analysis_fails(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
project_root = tmp_path / "project"
|
|
output_root = tmp_path / "workspace"
|
|
docs_root = output_root / "docs"
|
|
(project_root / "Flows").mkdir(parents=True)
|
|
(project_root / "Flows" / "Active.xaml").write_text(ACTIVE_XAML, encoding="utf-8")
|
|
(project_root / "Scripts").mkdir()
|
|
(project_root / "Scripts" / "Keep.bas").write_text("Sub Keep()\nEnd Sub", encoding="utf-8")
|
|
(project_root / "main.xaml").write_text(MAIN_XAML, encoding="utf-8")
|
|
|
|
report = ProjectPipeline(project_root, output_root, "main.xaml", force=True).run(FlakyAnalyzer())
|
|
|
|
self.assertTrue((docs_root / "Flows" / "Active.xaml.analysis.md").exists())
|
|
fallback = (docs_root / "Flows" / "Active.xaml.analysis.md").read_text(encoding="utf-8")
|
|
self.assertIn("Gemini 分析失败", fallback)
|
|
self.assertIn("HTTP 429 quota exhausted", fallback)
|
|
self.assertTrue((docs_root / "Scripts" / "Keep.bas.analysis.md").exists())
|
|
self.assertTrue(any("Analysis failed for Flows/Active.xaml" in item for item in report.warnings))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|