MCPcopy Create free account
hub / github.com/InfinitiBit/graphbit / _test_modular_workflows

Method _test_modular_workflows

scripts/legacy-migrator.py:253–302  ·  view source on GitHub ↗

Test modular workflows using the workflow tester.

(self)

Source from the content-addressed store, hash-verified

251 return False
252
253 def _test_modular_workflows(self) -> bool:
254 """Test modular workflows using the workflow tester."""
255 try:
256 # Run basic workflow tests
257 tester_script = self.root_path / "scripts" / "workflow-tester.py"
258
259 if not tester_script.exists():
260 print(" [WARN] Workflow tester not available, skipping comprehensive tests")
261 return True
262
263 # Get repository info from git
264 try:
265 git_path = shutil.which("git")
266 if not git_path:
267 raise FileNotFoundError("git executable not found in PATH")
268 repo_info = subprocess.run([git_path, "remote", "get-url", "origin"], capture_output=True, text=True, cwd=self.root_path, timeout=30, shell=False) # nosec
269
270 if repo_info.returncode == 0:
271 # Parse repo owner and name from URL
272 repo_url = repo_info.stdout.strip()
273 if "github.com" in repo_url:
274 # Extract owner/repo from URL
275 parts = repo_url.replace(".git", "").split("/")
276 if len(parts) >= 2:
277 repo_owner = parts[-2]
278 repo_name = parts[-1]
279
280 # Run preflight checks only (don't need full API access for migration)
281 print(f" [INFO] Detected repository: https://github.com/{repo_owner}/{repo_name}")
282 print(" [TEST] Running basic workflow validation...")
283
284 # Just validate that the tester can run
285 test_result = subprocess.run([sys.executable, str(tester_script), "--help"], capture_output=True, text=True, cwd=self.root_path, timeout=30, shell=False) # nosec
286
287 if test_result.returncode == 0:
288 print(" [PASS] Workflow tester functional")
289 return True
290 else:
291 print(" [FAIL] Workflow tester not functional")
292 return False
293
294 except Exception as e:
295 logger.warning(f"Failed to get repository info: {e}")
296
297 print(" [WARN] Could not determine repository info, skipping workflow tests")
298 return True
299
300 except Exception as e:
301 self.steps[3].error = str(e)
302 return False
303
304 def _update_branch_protection(self) -> bool:
305 """Update branch protection rules (optional step)."""

Callers 1

_execute_stepMethod · 0.95

Calls 1

runMethod · 0.80

Tested by

no test coverage detected