Merge pull request #1 from MariaHilmar/fix/wsl-git-coleta

fix: coleta Git via WSL (repos /root/MGI)

Author: MariaHilmar

coleta_git_contratos.py

@@ -6,68 +6,83 @@
 Consolidação de múltiplos repositórios na mesma saída
 """
 
+import os
 import subprocess
 import json
 from datetime import datetime
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple
 import re
 
+try:
+    from issue_keys import wsl_path_for_repo
+except ImportError:
+    def wsl_path_for_repo(repo: str) -> str:
+        paths = {
+            "contratos_v2": "/root/MGI/contratos_v2",
+            "contratos": "/root/MGI/contratos",
+        }
+        return paths.get(repo, "/root/MGI/contratos_v2")
+
+WSL_DISTRO = os.environ.get("MGI_WSL_DISTRO", "Ubuntu")
+
+
 class GitColeta:
     def __init__(self, repo_path: str, repo_name: Optional[str] = None):
         self.repo_path = Path(repo_path)
         self.repo_name = repo_name or Path(repo_path).name
+        self.wsl_repo_path = wsl_path_for_repo(self.repo_name)
         self.data: Dict = {
             'timestamp': datetime.now().isoformat(),
             'repositorio': self.repo_name,
             'caminho': str(repo_path),
+            'wsl_caminho': self.wsl_repo_path,
             'commits': [],
             'branches': [],
             'releases': [],
             'stats': {}
         }
 
-    def run_git(self, cmd: str, timeout: int = 30) -> str:
-        """Executa comando git no repositório.
-
-        Verifica o returncode e loga stderr para evitar falhas silenciosas.
-        """
+    def _run_git(self, git_args: str, timeout: int = 30) -> str:
+        """Executa git dentro do WSL Ubuntu (repos em /root/MGI/...)."""
+        cmd = [
+            "wsl",
+            "-d",
+            WSL_DISTRO,
+            "bash",
+            "-lc",
+            f"cd {self.wsl_repo_path} && git {git_args}",
+        ]
         try:
             result = subprocess.run(
                 cmd,
-                cwd=self.repo_path,
                 capture_output=True,
                 text=True,
-                shell=True,
                 timeout=timeout,
             )
             if result.returncode != 0:
-                stderr = (result.stderr or '').strip()
-                print(f"❌ Git retornou codigo {result.returncode} em {self.repo_name}: {stderr}")
+                stderr = (result.stderr or "").strip()
+                if stderr:
+                    print(f"ERRO Git ({self.repo_name}): {stderr}")
                 return ""
             return result.stdout.strip()
         except subprocess.TimeoutExpired:
-            print(f"❌ Timeout ({timeout}s) executando git em {self.repo_name}: {cmd}")
+            print(f"ERRO Timeout ({timeout}s) git em {self.repo_name}: {git_args[:80]}")
             return ""
         except Exception as e:
-            print(f"❌ Erro executando git em {self.repo_name}: {e}")
+            print(f"ERRO executando git em {self.repo_name}: {e}")
             return ""
 
+    def run_git(self, cmd: str, timeout: int = 30) -> str:
+        """Executa comando git no repositório (aceita 'git ...' ou args diretos)."""
+        git_args = cmd.strip()
+        if git_args.startswith("git "):
+            git_args = git_args[4:]
+        return self._run_git(git_args, timeout=timeout)
+
     def validar_repo(self) -> bool:
-        """Verifica se o repositório é acessível e é um repositório Git válido."""
-        try:
-            result = subprocess.run(
-                'git rev-parse --git-dir',
-                cwd=self.repo_path,
-                capture_output=True,
-                text=True,
-                shell=True,
-                timeout=10,
-            )
-            return result.returncode == 0
-        except Exception as e:
-            print(f"❌ Repositorio inacessivel {self.repo_name}: {e}")
-            return False
+        """Verifica se o repositório WSL é acessível e é um repositório Git válido."""
+        return bool(self._run_git("rev-parse --git-dir", timeout=10))
 
     def coleta_commits(self, since_days: int = 30) -> List[Dict[str, str]]:
         """Extrai commits dos últimos N dias"""

processar_issues_memoria.py

@@ -55,37 +55,12 @@ def probe_local_git_repos() -> bool:
 
 def probe_wsl_git_available(timeout_seconds: int = 8) -> bool:
     """Verifica se WSL Ubuntu + repo Git respondem (cache em memoria)."""
+    del timeout_seconds  # GitColeta valida via WSL com timeout proprio
     global _WSL_GIT_AVAILABLE
     if _WSL_GIT_AVAILABLE is not None:
         return _WSL_GIT_AVAILABLE
 
-    if not probe_local_git_repos():
-        _WSL_GIT_AVAILABLE = False
-        return False
-
-    import subprocess
-
-    from issue_keys import wsl_path_for_repo
-
-    repo_path = wsl_path_for_repo("contratos_v2")
-    cmd = [
-        "wsl",
-        "-d",
-        "Ubuntu",
-        "bash",
-        "-lc",
-        f"cd {repo_path} && git rev-parse --git-dir",
-    ]
-    try:
-        result = subprocess.run(
-            cmd,
-            capture_output=True,
-            text=True,
-            timeout=timeout_seconds,
-        )
-        _WSL_GIT_AVAILABLE = result.returncode == 0
-    except (OSError, subprocess.TimeoutExpired):
-        _WSL_GIT_AVAILABLE = False
+    _WSL_GIT_AVAILABLE = probe_local_git_repos()
     return _WSL_GIT_AVAILABLE
 
 

tests/test_coleta_git_contratos.py

@@ -0,0 +1,38 @@
+"""Testes da coleta Git via WSL."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import coleta_git_contratos as cgc
+
+
+def test_validar_repo_usa_wsl():
+    coleta = cgc.GitColeta("<path-contratos_v2>", "contratos_v2")
+    assert coleta.wsl_repo_path == "/root/MGI/contratos_v2"
+
+    with patch.object(coleta, "_run_git", return_value=".git") as mock_run:
+        assert coleta.validar_repo() is True
+        mock_run.assert_called_once_with("rev-parse --git-dir", timeout=10)
+
+
+def test_run_git_remove_prefixo_git():
+    coleta = cgc.GitColeta("<path-contratos>", "contratos")
+
+    with patch.object(coleta, "_run_git", return_value="ok") as mock_run:
+        assert coleta.run_git('git status -sb') == "ok"
+        mock_run.assert_called_once_with("status -sb", timeout=30)
+
+
+def test_run_git_via_wsl_comando():
+    coleta = cgc.GitColeta("<path-contratos_v2>", "contratos_v2")
+    mock_result = MagicMock(returncode=0, stdout="main\n", stderr="")
+
+    with patch("coleta_git_contratos.subprocess.run", return_value=mock_result) as mock_sub:
+        output = coleta._run_git("branch --show-current")
+
+    assert output == "main"
+    mock_sub.assert_called_once()
+    cmd = mock_sub.call_args.args[0]
+    assert cmd[:5] == ["wsl", "-d", "Ubuntu", "bash", "-lc"]
+    assert "cd /root/MGI/contratos_v2 && git branch --show-current" == cmd[5]