xiaomai
0cf62d1ac5
Reuse existing output directory by default instead of failing Cache successful Gemini analysis results using content hashing Skip unchanged files and retry failed analyses on subsequent runs Update --force flag to explicitly delete and rebuild the output
229 lines
11 KiB
Python
229 lines
11 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 RecordingAnalyzer:
|
|
def __init__(self) -> None:
|
|
self.paths: list[str] = []
|
|
|
|
def analyze(self, relative_path: Path, content: str) -> str:
|
|
self.paths.append(relative_path.as_posix())
|
|
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))
|
|
|
|
def test_pipeline_resume_skips_successfully_cached_analyses(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 / "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")
|
|
|
|
first = RecordingAnalyzer()
|
|
ProjectPipeline(project_root, output_root, "main.xaml", force=True).run(first)
|
|
self.assertEqual(
|
|
first.paths,
|
|
["Flows/Active.xaml", "main.xaml", "Scripts/Keep.bas"],
|
|
)
|
|
|
|
second = RecordingAnalyzer()
|
|
ProjectPipeline(project_root, output_root, "main.xaml", force=False).run(second)
|
|
self.assertEqual(second.paths, [])
|
|
|
|
def test_pipeline_resume_retries_failed_analysis_and_reanalyzes_changed_files(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 / "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")
|
|
|
|
ProjectPipeline(project_root, output_root, "main.xaml", force=True).run(FlakyAnalyzer())
|
|
|
|
(project_root / "Scripts" / "Keep.bas").write_text("Sub Keep()\nMsgBox \"updated\"\nEnd Sub", encoding="utf-8")
|
|
|
|
retry = RecordingAnalyzer()
|
|
ProjectPipeline(project_root, output_root, "main.xaml", force=False).run(retry)
|
|
self.assertEqual(retry.paths, ["Flows/Active.xaml", "Scripts/Keep.bas"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|