add all remote function

This commit is contained in:
VALLONGOL 2025-04-23 09:38:27 +02:00
parent 91698eac76
commit bebfb9a22a
6 changed files with 1828 additions and 375 deletions

File diff suppressed because it is too large Load Diff

159
ToDo.md
View File

@ -93,3 +93,162 @@ Basandomi sulle tue priorità e aggiungendo Fetch/Pull che sono complementari e
Questo piano costruisce le funzionalità in modo logico: prima ci si connette e si configura (`Apply`, `Check`), poi si sincronizza (`Fetch`, `Pull`, `Push`), poi si gestiscono casi speciali (`Clone`, `Push Tags`), e infine si aggiungono funzionalità di visualizzazione/confronto avanzate. Questo piano costruisce le funzionalità in modo logico: prima ci si connette e si configura (`Apply`, `Check`), poi si sincronizza (`Fetch`, `Pull`, `Push`), poi si gestiscono casi speciali (`Clone`, `Push Tags`), e infine si aggiungono funzionalità di visualizzazione/confronto avanzate.
Sei d'accordo con questo piano? Iniziamo con **Fetch**? Sei d'accordo con questo piano? Iniziamo con **Fetch**?
*STRATEGIA DI TEST*
Certamente! Testare con il tuo scenario specifico (repo remoto vuoto, repo locale "live") è un ottimo modo per verificare le nuove funzionalità. Ecco una guida passo passo:
**Scenario di Partenza:**
* **Locale:** Repository `GitUtility` con le ultime modifiche ai file `.py` per implementare Push/Pull, ma **non ancora committate**.
* **Remoto:** Repository corrispondente su Gitea (`http://172.23.34.96:3000/user/repo.git`) esistente ma **vuoto**.
* **Configurazione App:** Profilo caricato, URL e Nome remoto impostati nella tab "Remote Repository", premuto "Apply Config" e verificato che "Check Connection" dia "Connected (empty)".
**Guida al Test Passo Passo:**
**Passo 1: Test Pre-Commit (Verifica Blocchi)**
* **Azione:** Premi il pulsante **"Pull (Current Branch)"**.
* **Cosa Aspettarsi:**
* **Errore Atteso:** Dovresti ricevere immediatamente il popup "Pull Error" con il messaggio "Pull aborted: Uncommitted changes detected...".
* **Stato App:** L'operazione non dovrebbe partire, la status bar mostrerà l'errore, i pulsanti rimarranno abilitati.
* **Cosa Testiamo:** Verifica che il controllo pre-pull per le modifiche non committate funzioni correttamente.
* **Azione:** Premi il pulsante **"Push (Current Branch)"**.
* **Cosa Aspettarsi:**
* **Popup Atteso:** Dovrebbe apparire il popup "Uncommitted Changes" che chiede "Push anyway?".
* **Se premi No:** L'operazione si interrompe, la status bar dice "Push cancelled by user...".
* **Se premi Sì:** L'operazione continua (ma fallirà per un altro motivo tra poco).
* **Cosa Testiamo:** Verifica che il controllo pre-push (opzionale) funzioni e che l'utente venga avvisato. *Procedi premendo Sì per il prossimo test.*
* **Azione:** (Dopo aver premuto Sì al popup precedente) Attendi il completamento dell'operazione "Push".
* **Cosa Aspettarsi:**
* **Errore Atteso:** L'operazione dovrebbe fallire. Il popup di errore dovrebbe indicare qualcosa come "Push failed: Check branch upstream configuration..." o un errore simile legato al fatto che il branch remoto non esiste e l'upstream non è impostato (questo dipende da come `execute_remote_push` gestisce l'errore specifico del primo push senza `-u`, dato che il flag `force` era `False`). Potrebbe anche fallire per autenticazione se non l'avevi verificata.
* **Stato App:** Status bar rossa, indicatore auth potrebbe cambiare se l'errore era di autenticazione.
* **Cosa Testiamo:** Verifica che il push standard fallisca correttamente quando l'upstream non è impostato e il branch remoto non esiste.
**Passo 2: Commit Modifiche Locali**
* **Azione:**
1. Vai alla tab "Commit / Changes".
2. Premi "Refresh List" (dovresti vedere i file `.py` modificati).
3. Scrivi un messaggio di commit (es. "Implement Push Branch & Push Tags").
4. Premi "Commit Staged Changes".
* **Cosa Aspettarsi:**
* Operazione di commit asincrona parte e termina con successo.
* La lista "Working Directory Changes" si svuota.
* La status bar indica "Commit successful.".
* **Cosa Testiamo:** Funzionalità di commit locale standard (verifica che non sia stata compromessa).
**Passo 3: Test Primo Push (Impostazione Upstream)**
* **Azione:**
1. Torna alla tab "Remote Repository".
2. Assicurati che l'indicatore "Connection / Auth Status" sia verde ("Connected"). Se non lo è, premi "Check Connection".
3. Premi il pulsante **"Push (Current Branch)"**.
* **Cosa Aspettarsi:**
* **Operazione:** L'operazione asincrona parte. Il backend (`execute_remote_push`) dovrebbe rilevare che l'upstream non è impostato (`get_branch_upstream` restituisce `None`) e quindi chiamare `git_commands.git_push` con `set_upstream=True`.
* **Successo Atteso:** L'operazione dovrebbe terminare con successo. Git creerà il branch `master` (o il nome del tuo branch principale) sul remoto e imposterà il collegamento upstream.
* **GUI:** La status bar dovrebbe diventare verde con un messaggio tipo "Push branch 'master' to 'origin' completed successfully.". L'indicatore di connessione rimane verde.
* **Log:** Controlla i log per vedere il comando `git push --set-upstream origin master` (o simile) e l'output di successo.
* **Cosa Testiamo:** Funzionalità del primo push, creazione del branch remoto, impostazione automatica dell'upstream.
**Passo 4: Test Push Tags (Repo Remoto Non Vuoto)**
* **Azione:**
1. (Opzionale ma consigliato) Vai alla tab "Tags", crea un nuovo tag (es. `v.0.0.1.0` o simile) usando "Create New Tag...". Attendi il successo.
2. Torna alla tab "Remote Repository".
3. Premi il pulsante **"Push Tags"**.
4. Conferma nel popup `askyesno`.
* **Cosa Aspettarsi:**
* **Operazione:** Parte l'operazione asincrona che esegue `git push origin --tags`.
* **Successo Atteso:** L'operazione dovrebbe terminare con successo, inviando tutti i tag locali (incluso quello appena creato, se l'hai fatto) al remoto.
* **GUI:** Status bar verde con messaggio "Push tags to 'origin' completed successfully.".
* **Log:** Controlla i log per il comando `git push origin --tags`.
* **Verifica Remota (Opzionale):** Vai all'interfaccia web di Gitea e verifica che il tag (o i tag) siano apparsi nel repository remoto.
* **Cosa Testiamo:** Funzionalità di invio dei tag al repository remoto.
**Passo 5: Test Pull (Repo Remoto Non Vuoto, Nessuna Modifica)**
* **Azione:** Premi nuovamente il pulsante **"Pull (Current Branch)"**.
* **Cosa Aspettarsi:**
* **Operazione:** Parte l'operazione asincrona `git pull origin`.
* **Successo Atteso:** Dato che hai appena fatto push e non ci sono state modifiche remote, il pull dovrebbe terminare immediatamente con successo.
* **GUI:** Status bar verde con messaggio "Pull from 'origin': Repository already up-to-date.".
* **Cosa Testiamo:** Funzionamento del Pull quando non ci sono modifiche remote e l'upstream è configurato.
**Passo 6: Test Push (Nessuna Modifica Locale)**
* **Azione:** Premi nuovamente il pulsante **"Push (Current Branch)"**.
* **Cosa Aspettarsi:**
* **Operazione:** Parte l'operazione asincrona `git push origin master`.
* **Successo Atteso:** Dato che non ci sono nuovi commit locali, il push dovrebbe terminare immediatamente con successo.
* **GUI:** Status bar verde con messaggio "Push to 'origin': Branch 'master' already up-to-date.".
* **Cosa Testiamo:** Funzionamento del Push quando non ci sono modifiche locali da inviare.
**Passo 7: Test Fetch (Nessuna Modifica)**
* **Azione:** Premi il pulsante **"Fetch"**.
* **Cosa Aspettarsi:**
* **Operazione:** Parte l'operazione `git fetch origin --prune`.
* **Successo Atteso:** Termina con successo.
* **GUI:** Status bar verde con messaggio "Fetch from 'origin' completed successfully.". L'output potrebbe essere minimo o nullo se non c'erano branch remoti da "prunare".
* **Cosa Testiamo:** Funzionamento base del Fetch.
**Passo 8: Test Rifiuto Push (Simulazione Modifica Remota)**
* **Azione (Esterna):**
1. Vai all'interfaccia web di Gitea.
2. Naviga nel tuo repository.
3. Trova un modo per modificare un file direttamente da Gitea (es. modifica il README se esiste, o crea/modifica un file di testo) e fai il commit di questa modifica **direttamente sul server nel branch `master`**.
* **Azione (App):**
1. Nell'applicazione GitUtility, **NON FARE Fetch o Pull**.
2. Fai una piccola modifica locale a un file qualsiasi (es. aggiungi un commento).
3. Vai alla tab "Commit / Changes", fai il commit di questa modifica locale ("Test commit for rejection").
4. Torna alla tab "Remote Repository".
5. Premi **"Push (Current Branch)"**.
* **Cosa Aspettarsi:**
* **Operazione:** Parte `git push origin master`.
* **Fallimento Atteso:** Il push dovrebbe essere **rifiutato** perché il repository remoto (`origin/master`) ora contiene un commit (quello fatto via web) che il tuo repository locale non ha. È un errore "non-fast-forward".
* **GUI:**
* La status bar diventa rossa con lo stato `'rejected'`.
* Dovrebbe apparire un popup `messagebox.showwarning("Push Rejected", ...)` con il messaggio che spiega il rifiuto e suggerisce di fare pull.
* **Log:** Verifica che venga loggato l'errore "Push rejected (non-fast-forward)".
* **Cosa Testiamo:** Rilevamento e gestione corretta del rifiuto di push dovuto a divergenza tra locale e remoto.
**Passo 9: Test Risoluzione Rifiuto (Pull con Conflitto o Merge)**
* **Azione:** Dopo il push rifiutato nel passo 8, premi **"Pull (Current Branch)"**.
* **Cosa Aspettarsi:**
* **Operazione:** Parte `git pull origin`. Git tenterà di fare il fetch del commit remoto e poi il merge nel tuo branch locale `master`.
* **Esito Possibile 1 (Conflitto):** Se la modifica che hai fatto localmente e quella fatta su Gitea riguardano la stessa parte dello stesso file, si verificherà un **conflitto di merge**.
* **GUI:** Status bar rossa con stato `'conflict'`, popup "Merge Conflict" che ti dice di risolvere manualmente i conflitti nel path specificato. La lista "Working Directory Changes" dovrebbe mostrare il file in conflitto con uno stato tipo `UU`.
* **Esito Possibile 2 (Merge Riuscito):** Se le modifiche erano su file diversi o parti diverse dello stesso file, Git potrebbe riuscire a fare il merge automaticamente.
* **GUI:** Status bar verde con "Pull from 'origin' completed successfully.". Potrebbe aprirsi un editor di testo per il messaggio di merge commit (se Git è configurato per farlo - la nostra app non lo gestisce, quindi probabilmente il merge commit avrà un messaggio standard). La lista "Working Directory Changes" dovrebbe essere vuota (a meno che il merge non abbia lasciato file non risolti, improbabile per merge automatico).
* **Cosa Testiamo:** Capacità di rilevare e segnalare conflitti di merge durante il Pull, o gestione del merge automatico riuscito.
**Passo 10 (Se C'è Stato Conflitto): Risoluzione Manuale e Push Finale**
* **Azione (Esterna):**
1. Apri il file in conflitto indicato nel messaggio con un editor di testo.
2. Risolvi manualmente i marcatori di conflitto (`<<<<<<<`, `=======`, `>>>>>>>`).
3. Salva il file.
* **Azione (App):**
1. Vai alla tab "Commit / Changes".
2. Premi "Refresh List". Dovresti vedere il file precedentemente in conflitto ora come modificato (`M`).
3. Seleziona il file risolto e usa il menu contestuale (tasto destro) per fare "Add to Staging Area" (o usa `git add <file>` da terminale). **Nota:** Dobbiamo assicurarci che il menu contestuale funzioni anche per i file modificati, non solo per quelli untracked, per poterli aggiungere. Potrebbe essere più semplice fare il commit direttamente.
4. Scrivi un messaggio di commit (es. "Merge remote changes after resolving conflict").
5. Premi "Commit Staged Changes".
6. Torna alla tab "Remote Repository".
7. Premi **"Push (Current Branch)"**.
* **Cosa Aspettarsi:**
* **Operazione:** Parte `git push origin master`.
* **Successo Atteso:** Questa volta il push dovrebbe riuscire perché il tuo branch locale ora include sia le tue modifiche sia quelle remote (integrate dal merge commit).
* **GUI:** Status bar verde "Push branch 'master' to 'origin' completed successfully.".
* **Cosa Testiamo:** Flusso completo di risoluzione conflitto (anche se la risoluzione è manuale) e push successivo.
Questo piano di test copre i principali scenari per le funzionalità remote implementate. Eseguilo con attenzione e controlla i log e il comportamento della GUI ad ogni passo. Fammi sapere se qualcosa non funziona come previsto!

View File

@ -964,26 +964,36 @@ def run_pull_remote_async(
repo_path: str, repo_path: str,
remote_name: str, remote_name: str,
# branch_name non è più necessario passarlo qui, lo otteniamo nel worker # branch_name non è più necessario passarlo qui, lo otteniamo nel worker
results_queue: queue.Queue results_queue: queue.Queue,
): ):
""" """
Worker function to execute 'git pull' asynchronously. Worker function to execute 'git pull' asynchronously.
Executed in a separate thread. Executed in a separate thread.
""" """
func_name = "run_pull_remote_async" func_name = "run_pull_remote_async"
log_handler.log_debug(f"[Worker] Started: Pull Remote '{remote_name}' for '{repo_path}'", func_name=func_name) log_handler.log_debug(
f"[Worker] Started: Pull Remote '{remote_name}' for '{repo_path}'",
func_name=func_name,
)
try: try:
# --- Ottieni il branch corrente --- # --- Ottieni il branch corrente ---
# È necessario conoscere il branch corrente per passarlo a execute_remote_pull # È necessario conoscere il branch corrente per passarlo a execute_remote_pull
# (anche se git pull di base non lo richiede, la nostra logica di action lo usa) # (anche se git pull di base non lo richiede, la nostra logica di action lo usa)
# e per log/messaggi più chiari. # e per log/messaggi più chiari.
current_branch_name = git_commands.get_current_branch_name(repo_path) # Assumendo che questo metodo esista/venga aggiunto a GitCommands current_branch_name = git_commands.get_current_branch_name(
repo_path
) # Assumendo che questo metodo esista/venga aggiunto a GitCommands
if not current_branch_name: if not current_branch_name:
# Se non riusciamo a determinare il branch (es. detached HEAD), non possiamo fare pull standard # Se non riusciamo a determinare il branch (es. detached HEAD), non possiamo fare pull standard
raise ValueError("Cannot perform pull: Unable to determine current branch (possibly detached HEAD).") raise ValueError(
"Cannot perform pull: Unable to determine current branch (possibly detached HEAD)."
)
log_handler.log_debug(f"[Worker] Current branch identified as: '{current_branch_name}'", func_name=func_name) log_handler.log_debug(
f"[Worker] Current branch identified as: '{current_branch_name}'",
func_name=func_name,
)
# --- Chiama l'Action Handler --- # --- Chiama l'Action Handler ---
# execute_remote_pull contiene la logica per: # execute_remote_pull contiene la logica per:
@ -995,18 +1005,26 @@ def run_pull_remote_async(
) )
# result_info è il dizionario restituito da execute_remote_pull # result_info è il dizionario restituito da execute_remote_pull
log_handler.log_info(f"[Worker] Pull result status for '{remote_name}': {result_info.get('status')}", func_name=func_name) log_handler.log_info(
f"[Worker] Pull result status for '{remote_name}': {result_info.get('status')}",
func_name=func_name,
)
# --- Aggiungi informazioni extra per la gestione dei conflitti --- # --- Aggiungi informazioni extra per la gestione dei conflitti ---
if result_info.get('status') == 'conflict': if result_info.get("status") == "conflict":
result_info['repo_path'] = repo_path # Assicura che il path sia nel risultato per il messaggio GUI result_info["repo_path"] = (
repo_path # Assicura che il path sia nel risultato per il messaggio GUI
)
# Metti il dizionario del risultato direttamente nella coda # Metti il dizionario del risultato direttamente nella coda
results_queue.put(result_info) results_queue.put(result_info)
except (GitCommandError, ValueError) as e: except (GitCommandError, ValueError) as e:
# Cattura errori dalla determinazione del branch o altri errori noti # Cattura errori dalla determinazione del branch o altri errori noti
log_handler.log_error(f"[Worker] Handled EXCEPTION during pull setup/execution: {e}", func_name=func_name) log_handler.log_error(
f"[Worker] Handled EXCEPTION during pull setup/execution: {e}",
func_name=func_name,
)
results_queue.put( results_queue.put(
{ {
"status": "error", "status": "error",
@ -1016,7 +1034,10 @@ def run_pull_remote_async(
) )
except Exception as e: except Exception as e:
# Cattura eccezioni impreviste # Cattura eccezioni impreviste
log_handler.log_exception(f"[Worker] UNEXPECTED EXCEPTION during pull operation: {e}", func_name=func_name) log_handler.log_exception(
f"[Worker] UNEXPECTED EXCEPTION during pull operation: {e}",
func_name=func_name,
)
results_queue.put( results_queue.put(
{ {
"status": "error", "status": "error",
@ -1025,8 +1046,163 @@ def run_pull_remote_async(
} }
) )
finally: finally:
log_handler.log_debug(f"[Worker] Finished: Pull Remote '{remote_name}'", func_name=func_name) log_handler.log_debug(
# def run_push_remote_async(remote_action_handler, repo_path, remote_name, branch_name, results_queue): pass f"[Worker] Finished: Pull Remote '{remote_name}'", func_name=func_name
# def run_push_tags_async(remote_action_handler, repo_path, remote_name, results_queue): pass )
def run_push_remote_async(
remote_action_handler: RemoteActionHandler, # Dipendenza per eseguire push
git_commands: GitCommands, # Necessario per ottenere il branch corrente
repo_path: str,
remote_name: str,
# branch_name viene determinato internamente
# force flag non necessario qui, gestito da execute_remote_push se necessario
results_queue: queue.Queue,
):
"""
Worker function to execute 'git push' for the current branch asynchronously.
Executed in a separate thread. Handles upstream setting.
"""
func_name = "run_push_remote_async"
log_handler.log_debug(
f"[Worker] Started: Push Remote '{remote_name}' for '{repo_path}'",
func_name=func_name,
)
try:
# --- Ottieni il branch corrente ---
current_branch_name = git_commands.get_current_branch_name(repo_path)
if not current_branch_name:
raise ValueError(
"Cannot perform push: Unable to determine current branch (possibly detached HEAD)."
)
log_handler.log_debug(
f"[Worker] Current branch identified as: '{current_branch_name}'",
func_name=func_name,
)
# --- Chiama l'Action Handler ---
# execute_remote_push contiene la logica per:
# 1. Controllare e impostare l'upstream se necessario
# 2. Eseguire git push
# 3. Analizzare l'esito (successo, rifiutato, errore auth/conn/altro)
# NOTA: Il flag 'force' qui è sempre False per il push standard dal pulsante GUI.
# Un eventuale push forzato richiederebbe un pulsante/conferma separata.
result_info = remote_action_handler.execute_remote_push(
repo_path=repo_path,
remote_name=remote_name,
current_branch_name=current_branch_name,
force=False, # Push standard
)
# result_info è il dizionario restituito da execute_remote_push
log_handler.log_info(
f"[Worker] Push result status for '{current_branch_name}' to '{remote_name}': {result_info.get('status')}",
func_name=func_name,
)
# --- Aggiungi informazioni extra per la gestione GUI ---
if result_info.get("status") == "rejected":
result_info["branch_name"] = (
current_branch_name # Passa il nome del branch nel risultato
)
# Metti il dizionario del risultato direttamente nella coda
results_queue.put(result_info)
except (GitCommandError, ValueError) as e:
# Cattura errori dalla determinazione del branch o altri errori noti
log_handler.log_error(
f"[Worker] Handled EXCEPTION during push setup/execution: {e}",
func_name=func_name,
)
results_queue.put(
{
"status": "error",
"exception": e,
"message": f"Push failed: {e}", # Usa messaggio eccezione
}
)
except Exception as e:
# Cattura eccezioni impreviste
log_handler.log_exception(
f"[Worker] UNEXPECTED EXCEPTION during push operation: {e}",
func_name=func_name,
)
results_queue.put(
{
"status": "error",
"exception": e,
"message": f"Unexpected error during push operation: {type(e).__name__}",
}
)
finally:
log_handler.log_debug(
f"[Worker] Finished: Push Remote '{remote_name}'", func_name=func_name
)
def run_push_tags_async(
remote_action_handler: RemoteActionHandler, # Dipendenza per eseguire push tags
repo_path: str,
remote_name: str,
results_queue: queue.Queue,
):
"""
Worker function to execute 'git push --tags' asynchronously.
Executed in a separate thread.
"""
func_name = "run_push_tags_async"
log_handler.log_debug(
f"[Worker] Started: Push Tags to '{remote_name}' for '{repo_path}'",
func_name=func_name,
)
try:
# --- Chiama l'Action Handler ---
result_info = remote_action_handler.execute_push_tags(repo_path, remote_name)
# result_info è il dizionario restituito da execute_push_tags
log_handler.log_info(
f"[Worker] Push tags result status for '{remote_name}': {result_info.get('status')}",
func_name=func_name,
)
# Metti il dizionario del risultato direttamente nella coda
results_queue.put(result_info)
except (GitCommandError, ValueError) as e:
# Cattura errori noti sollevati da execute_push_tags (principalmente validazione)
log_handler.log_error(
f"[Worker] Handled EXCEPTION during push tags execution: {e}",
func_name=func_name,
)
results_queue.put(
{
"status": "error",
"exception": e,
"message": f"Push tags failed: {e}",
}
)
except Exception as e:
# Cattura eccezioni impreviste
log_handler.log_exception(
f"[Worker] UNEXPECTED EXCEPTION during push tags operation: {e}",
func_name=func_name,
)
results_queue.put(
{
"status": "error",
"exception": e,
"message": f"Unexpected error during push tags operation: {type(e).__name__}",
}
)
finally:
log_handler.log_debug(
f"[Worker] Finished: Push Tags to '{remote_name}'", func_name=func_name
)
# --- END OF FILE async_workers.py --- # --- END OF FILE async_workers.py ---

View File

@ -1359,7 +1359,9 @@ class GitCommands:
) )
return result return result
def git_pull(self, working_directory: str, remote_name: str, branch_name: str) -> subprocess.CompletedProcess: def git_pull(
self, working_directory: str, remote_name: str, branch_name: str
) -> subprocess.CompletedProcess:
""" """
Executes 'git pull <remote_name> <branch_name>'. Executes 'git pull <remote_name> <branch_name>'.
This performs a fetch and then merges the fetched branch into the current local branch. This performs a fetch and then merges the fetched branch into the current local branch.
@ -1388,7 +1390,7 @@ class GitCommands:
log_handler.log_info( log_handler.log_info(
f"Pulling from remote '{remote_name}' into current branch ('{branch_name}') " f"Pulling from remote '{remote_name}' into current branch ('{branch_name}') "
f"in '{working_directory}'", f"in '{working_directory}'",
func_name=func_name func_name=func_name,
) )
cmd = ["git", "pull", remote_name] cmd = ["git", "pull", remote_name]
@ -1400,7 +1402,7 @@ class GitCommands:
check=False, # Fondamentale per rilevare conflitti (RC=1) check=False, # Fondamentale per rilevare conflitti (RC=1)
capture=True, capture=True,
hide_console=True, hide_console=True,
log_output_level=logging.INFO # Logga output (aggiornamenti, merge) a INFO log_output_level=logging.INFO, # Logga output (aggiornamenti, merge) a INFO
) )
return result return result
@ -1409,14 +1411,93 @@ class GitCommands:
working_directory: str, working_directory: str,
remote_name: str, remote_name: str,
branch_name: str, branch_name: str,
set_upstream=False, set_upstream: bool = False,
): force: bool = False, # Aggiunto parametro opzionale per force push
# To be implemented: git push [--set-upstream] <remote_name> <branch_name> ) -> subprocess.CompletedProcess:
pass """
Executes 'git push [<options>] <remote_name> <branch_name>'.
Handles setting upstream and optional force push.
Does NOT raise exception on non-zero exit code (check=False).
def git_push_tags(self, working_directory: str, remote_name: str): Args:
# To be implemented: git push <remote_name> --tags working_directory (str): Path to the repository.
pass remote_name (str): The name of the remote to push to.
branch_name (str): The name of the local branch to push.
set_upstream (bool): If True, add '-u' or '--set-upstream' flag.
force (bool): If True, add '--force' flag (use with extreme caution!).
Returns:
subprocess.CompletedProcess: The result of the command execution.
"""
func_name = "git_push"
push_options = []
if set_upstream:
push_options.append("--set-upstream")
log_handler.log_info(
f"Pushing branch '{branch_name}' to '{remote_name}' and setting upstream.",
func_name=func_name,
)
else:
log_handler.log_info(
f"Pushing branch '{branch_name}' to '{remote_name}'.",
func_name=func_name,
)
if force:
push_options.append("--force")
log_handler.log_warning(
f"Executing FORCE PUSH for branch '{branch_name}' to '{remote_name}'!",
func_name=func_name,
)
# Comando base: git push [options] <remote> <local_branch>[:<remote_branch>]
# Specificare il refspec completo (<local>:<remote>) è più robusto,
# specialmente se i nomi non coincidono. Per ora, assumiamo che coincidano.
cmd = ["git", "push"] + push_options + [remote_name, branch_name]
# Esegui catturando output, nascondendo console, ma NON sollevare eccezione (check=False)
# Il worker analizzerà il risultato per errori specifici (rifiutato, auth, etc.)
result = self.log_and_execute(
command=cmd,
working_directory=working_directory,
check=False, # Importante per rilevare push rifiutati (RC=1)
capture=True,
hide_console=True,
log_output_level=logging.INFO, # Logga output (es. riepilogo push) a INFO
)
return result
def git_push_tags(
self, working_directory: str, remote_name: str
) -> subprocess.CompletedProcess:
"""
Executes 'git push <remote_name> --tags'.
Does NOT raise exception on non-zero exit code (check=False).
Args:
working_directory (str): Path to the repository.
remote_name (str): The name of the remote to push tags to.
Returns:
subprocess.CompletedProcess: The result of the command execution.
"""
func_name = "git_push_tags"
log_handler.log_info(
f"Pushing all tags to remote '{remote_name}' from '{working_directory}'",
func_name=func_name,
)
cmd = ["git", "push", remote_name, "--tags"]
# Esegui catturando output, nascondendo console, ma NON sollevare eccezione (check=False)
result = self.log_and_execute(
command=cmd,
working_directory=working_directory,
check=False,
capture=True,
hide_console=True,
log_output_level=logging.INFO,
)
return result
def get_current_branch_name(self, working_directory: str) -> str | None: def get_current_branch_name(self, working_directory: str) -> str | None:
""" """
@ -1424,29 +1505,156 @@ class GitCommands:
Returns None if in detached HEAD state or on error. Returns None if in detached HEAD state or on error.
""" """
func_name = "get_current_branch_name" func_name = "get_current_branch_name"
log_handler.log_debug(f"Getting current branch name in '{working_directory}'", func_name=func_name) log_handler.log_debug(
f"Getting current branch name in '{working_directory}'", func_name=func_name
)
# Usa 'git branch --show-current' (disponibile da Git 2.22+) # Usa 'git branch --show-current' (disponibile da Git 2.22+)
# In alternativa 'git rev-parse --abbrev-ref HEAD', che funziona anche prima # In alternativa 'git rev-parse --abbrev-ref HEAD', che funziona anche prima
# ma può restituire 'HEAD' in detached state. 'symbolic-ref' è più robusto. # ma può restituire 'HEAD' in detached state. 'symbolic-ref' è più robusto.
cmd = ["git", "symbolic-ref", "--short", "-q", "HEAD"] # -q sopprime errori se non è un branch (detached) cmd = [
"git",
"symbolic-ref",
"--short",
"-q",
"HEAD",
] # -q sopprime errori se non è un branch (detached)
try: try:
# check=False perché può fallire legittimamente in detached HEAD (RC=1) # check=False perché può fallire legittimamente in detached HEAD (RC=1)
result = self.log_and_execute(cmd, working_directory, check=False, capture=True, hide_console=True, log_output_level=logging.DEBUG) result = self.log_and_execute(
cmd,
working_directory,
check=False,
capture=True,
hide_console=True,
log_output_level=logging.DEBUG,
)
if result.returncode == 0 and result.stdout: if result.returncode == 0 and result.stdout:
branch_name = result.stdout.strip() branch_name = result.stdout.strip()
log_handler.log_info(f"Current branch is '{branch_name}'", func_name=func_name) log_handler.log_info(
f"Current branch is '{branch_name}'", func_name=func_name
)
return branch_name return branch_name
elif result.returncode == 1: # Codice atteso per detached HEAD con symbolic-ref -q elif (
log_handler.log_warning("Currently in detached HEAD state.", func_name=func_name) result.returncode == 1
): # Codice atteso per detached HEAD con symbolic-ref -q
log_handler.log_warning(
"Currently in detached HEAD state.", func_name=func_name
)
return None return None
else: # Altro errore else: # Altro errore
log_handler.log_error(f"Failed to get current branch (RC={result.returncode}). Stderr: {result.stderr.strip() if result.stderr else 'N/A'}", func_name=func_name) log_handler.log_error(
f"Failed to get current branch (RC={result.returncode}). Stderr: {result.stderr.strip() if result.stderr else 'N/A'}",
func_name=func_name,
)
return None return None
except Exception as e: except Exception as e:
log_handler.log_exception(f"Unexpected error getting current branch: {e}", func_name=func_name) log_handler.log_exception(
f"Unexpected error getting current branch: {e}", func_name=func_name
)
return None # Ritorna None anche per eccezioni impreviste return None # Ritorna None anche per eccezioni impreviste
def get_branch_upstream(
self, working_directory: str, branch_name: str
) -> str | None:
"""
Gets the upstream remote branch configured for a local branch.
Args:
working_directory (str): Path to the repository.
branch_name (str): The name of the local branch.
Returns:
str | None: The full name of the upstream branch (e.g., 'origin/main')
or None if no upstream is configured or on error.
"""
func_name = "get_branch_upstream"
log_handler.log_debug(
f"Getting upstream for branch '{branch_name}' in '{working_directory}'",
func_name=func_name,
)
# Usa 'git rev-parse --abbrev-ref <branch>@{upstream}'
# Questo comando restituisce il nome breve dell'upstream se esiste, altrimenti fallisce.
# L'uso di @{upstream} è un modo standard per riferirsi all'upstream configurato.
cmd = ["git", "rev-parse", "--abbrev-ref", f"{branch_name}@{{upstream}}"]
try:
# Esegui catturando output, nascondendo console. check=False perché fallisce se non c'è upstream.
result = self.log_and_execute(
cmd,
working_directory,
check=False,
capture=True,
hide_console=True,
log_output_level=logging.DEBUG,
)
if result.returncode == 0 and result.stdout:
upstream_name = result.stdout.strip()
# Il comando sopra restituisce solo il nome del branch remoto (es. 'main'),
# ma noi vogliamo il nome completo 'remote/branch'. Dobbiamo costruirlo.
# Possiamo usare `git config branch.<name>.remote` e `branch.<name>.merge`.
remote_cmd = ["git", "config", "--get", f"branch.{branch_name}.remote"]
merge_ref_cmd = [
"git",
"config",
"--get",
f"branch.{branch_name}.merge",
]
remote_result = self.log_and_execute(
remote_cmd,
working_directory,
check=False,
capture=True,
hide_console=True,
)
merge_ref_result = self.log_and_execute(
merge_ref_cmd,
working_directory,
check=False,
capture=True,
hide_console=True,
)
if remote_result.returncode == 0 and merge_ref_result.returncode == 0:
remote = remote_result.stdout.strip()
merge_ref = merge_ref_result.stdout.strip() # Es. 'refs/heads/main'
# Estrai il nome semplice del branch dal ref completo
remote_branch_name = merge_ref.split("/")[-1]
full_upstream_name = f"{remote}/{remote_branch_name}"
log_handler.log_info(
f"Upstream for '{branch_name}' is '{full_upstream_name}'",
func_name=func_name,
)
return full_upstream_name
else:
# Se non riusciamo a ottenere remote/merge, c'è un problema di configurazione
log_handler.log_warning(
f"Could not determine full upstream name for '{branch_name}' despite rev-parse success.",
func_name=func_name,
)
return None # O forse l'upstream_name da rev-parse? Meglio essere sicuri.
elif "no upstream configured" in (result.stderr or "").lower():
log_handler.log_info(
f"No upstream configured for branch '{branch_name}'.",
func_name=func_name,
)
return None
else: # Altro errore da rev-parse
log_handler.log_error(
f"Failed to get upstream for '{branch_name}' (RC={result.returncode}). Stderr: {result.stderr.strip() if result.stderr else 'N/A'}",
func_name=func_name,
)
return None
except Exception as e:
log_handler.log_exception(
f"Unexpected error getting upstream for '{branch_name}': {e}",
func_name=func_name,
)
return None
# --- END OF FILE git_commands.py --- # --- END OF FILE git_commands.py ---

52
gui.py
View File

@ -446,8 +446,8 @@ class MainFrame(ttk.Frame):
check_connection_auth_cb, check_connection_auth_cb,
fetch_remote_cb, fetch_remote_cb,
pull_remote_cb, pull_remote_cb,
# push_remote_cb, # Placeholder per futuro push_remote_cb,
# push_tags_remote_cb, # Placeholder per futuro push_tags_remote_cb,
): ):
"""Initializes the MainFrame.""" """Initializes the MainFrame."""
super().__init__(master) super().__init__(master)
@ -481,8 +481,8 @@ class MainFrame(ttk.Frame):
self.check_connection_auth_callback = check_connection_auth_cb self.check_connection_auth_callback = check_connection_auth_cb
self.fetch_remote_callback = fetch_remote_cb self.fetch_remote_callback = fetch_remote_cb
self.pull_remote_callback = pull_remote_cb self.pull_remote_callback = pull_remote_cb
# self.push_remote_callback = push_remote_cb self.push_remote_callback = push_remote_cb
# self.push_tags_remote_callback = push_tags_remote_cb self.push_tags_remote_callback = push_tags_remote_cb
# Configure style (invariato) # Configure style (invariato)
self.style = ttk.Style() self.style = ttk.Style()
@ -721,17 +721,39 @@ class MainFrame(ttk.Frame):
actions_frame, actions_frame,
text="Pull (Current Branch)", text="Pull (Current Branch)",
command=self.pull_remote_callback, # Usa il nuovo callback command=self.pull_remote_callback, # Usa il nuovo callback
state=tk.DISABLED # Abilitato quando repo pronto e connesso? state=tk.DISABLED, # Abilitato quando repo pronto e connesso?
) )
self.pull_button.pack(side=tk.LEFT, padx=5, pady=5) self.pull_button.pack(side=tk.LEFT, padx=5, pady=5)
self.create_tooltip(self.pull_button, "Fetch from and integrate with the remote branch corresponding to the current local branch (merge or rebase).") self.create_tooltip(
# self.create_tooltip(self.pull_button, ...) self.pull_button,
# self.push_button = ttk.Button(actions_frame, text="Push (Current Branch)", command=self.push_remote_callback, state=tk.DISABLED) "Fetch from and integrate with the remote branch corresponding to the current local branch (merge or rebase).",
# self.push_button.pack(side=tk.LEFT, padx=5, pady=5) )
# self.create_tooltip(self.push_button, ...)
# self.push_tags_button = ttk.Button(actions_frame, text="Push Tags", command=self.push_tags_remote_callback, state=tk.DISABLED) self.push_button = ttk.Button(
# self.push_tags_button.pack(side=tk.LEFT, padx=5, pady=5) actions_frame,
# self.create_tooltip(self.push_tags_button, ...) text="Push (Current Branch)",
command=self.push_remote_callback, # Nuovo callback
state=tk.DISABLED, # Abilitato quando repo pronto e connesso?
)
self.push_button.pack(side=tk.LEFT, padx=5, pady=5)
self.create_tooltip(
self.push_button,
"Upload local branch commits to the corresponding remote branch. Will set upstream on first push.",
)
# ---<<< FINE NUOVO PULSANTE PUSH BRANCH >>>---
# ---<<< NUOVO PULSANTE PUSH TAGS >>>---
self.push_tags_button = ttk.Button(
actions_frame,
text="Push Tags",
command=self.push_tags_remote_callback, # Nuovo callback
state=tk.DISABLED, # Abilitato quando repo pronto e connesso?
)
self.push_tags_button.pack(side=tk.LEFT, padx=5, pady=5)
self.create_tooltip(
self.push_tags_button,
"Upload all local tags (created via the Tags tab) to the remote repository.",
)
return frame return frame
@ -1838,8 +1860,8 @@ class MainFrame(ttk.Frame):
self.check_auth_button, self.check_auth_button,
self.fetch_button, self.fetch_button,
self.pull_button, self.pull_button,
# self.push_button, self.push_button,
# self.push_tags_button, self.push_tags_button,
] ]
# log_handler.log_debug(f"Setting action widgets state to: {state}", func_name="set_action_widgets_state") # Usa log_handler # log_handler.log_debug(f"Setting action widgets state to: {state}", func_name="set_action_widgets_state") # Usa log_handler
for widget in widgets: for widget in widgets:

View File

@ -142,8 +142,6 @@ class RemoteActionHandler:
# Wrap unexpected errors in GitCommandError or a custom one if needed # Wrap unexpected errors in GitCommandError or a custom one if needed
raise GitCommandError(f"Unexpected error: {e}") from e raise GitCommandError(f"Unexpected error: {e}") from e
# --- Placeholder for future remote methods ---
def execute_remote_fetch(self, repo_path: str, remote_name: str) -> dict: def execute_remote_fetch(self, repo_path: str, remote_name: str) -> dict:
""" """
Executes 'git fetch' for the specified remote. Executes 'git fetch' for the specified remote.
@ -277,7 +275,9 @@ class RemoteActionHandler:
return result_info return result_info
def execute_remote_pull(self, repo_path: str, remote_name: str, current_branch_name: str) -> dict: def execute_remote_pull(
self, repo_path: str, remote_name: str, current_branch_name: str
) -> dict:
""" """
Executes 'git pull' for the specified remote and current branch. Executes 'git pull' for the specified remote and current branch.
Detects success, merge conflicts, and other errors. Detects success, merge conflicts, and other errors.
@ -300,7 +300,7 @@ class RemoteActionHandler:
func_name = "execute_remote_pull" func_name = "execute_remote_pull"
log_handler.log_info( log_handler.log_info(
f"Executing pull from remote '{remote_name}' into branch '{current_branch_name}' in '{repo_path}'", f"Executing pull from remote '{remote_name}' into branch '{current_branch_name}' in '{repo_path}'",
func_name=func_name func_name=func_name,
) )
# --- Input Validation --- # --- Input Validation ---
@ -321,18 +321,20 @@ class RemoteActionHandler:
msg = "Pull aborted: Uncommitted changes detected in the working directory. Please commit or stash first." msg = "Pull aborted: Uncommitted changes detected in the working directory. Please commit or stash first."
log_handler.log_warning(msg, func_name=func_name) log_handler.log_warning(msg, func_name=func_name)
# Restituisce un errore specifico per questo caso # Restituisce un errore specifico per questo caso
return {'status': 'error', 'message': msg, 'exception': ValueError(msg)} return {"status": "error", "message": msg, "exception": ValueError(msg)}
except GitCommandError as status_err: except GitCommandError as status_err:
# Se il controllo dello stato fallisce, non procedere # Se il controllo dello stato fallisce, non procedere
msg = f"Pull aborted: Failed to check repository status before pull: {status_err}" msg = f"Pull aborted: Failed to check repository status before pull: {status_err}"
log_handler.log_error(msg, func_name=func_name) log_handler.log_error(msg, func_name=func_name)
return {'status': 'error', 'message': msg, 'exception': status_err} return {"status": "error", "message": msg, "exception": status_err}
# --- Esecuzione Git Pull --- # --- Esecuzione Git Pull ---
result_info = {'status': 'unknown', 'message': 'Pull not completed.'} # Default result_info = {"status": "unknown", "message": "Pull not completed."} # Default
try: try:
# Chiama il metodo git_pull (che ha check=False) # Chiama il metodo git_pull (che ha check=False)
pull_result = self.git_commands.git_pull(repo_path, remote_name, current_branch_name) pull_result = self.git_commands.git_pull(
repo_path, remote_name, current_branch_name
)
# Analizza il risultato del comando pull # Analizza il risultato del comando pull
stdout_full = pull_result.stdout if pull_result.stdout else "" stdout_full = pull_result.stdout if pull_result.stdout else ""
@ -341,70 +343,494 @@ class RemoteActionHandler:
if pull_result.returncode == 0: if pull_result.returncode == 0:
# Successo # Successo
result_info['status'] = 'success' result_info["status"] = "success"
if "already up to date" in combined_output_lower: if "already up to date" in combined_output_lower:
result_info['message'] = f"Pull from '{remote_name}': Repository already up-to-date." result_info["message"] = (
log_handler.log_info(f"Pull successful (already up-to-date) for '{remote_name}'.", func_name=func_name) f"Pull from '{remote_name}': Repository already up-to-date."
)
log_handler.log_info(
f"Pull successful (already up-to-date) for '{remote_name}'.",
func_name=func_name,
)
else: else:
result_info['message'] = f"Pull from '{remote_name}' completed successfully." result_info["message"] = (
log_handler.log_info(f"Pull successful for '{remote_name}'. Output logged.", func_name=func_name) f"Pull from '{remote_name}' completed successfully."
)
log_handler.log_info(
f"Pull successful for '{remote_name}'. Output logged.",
func_name=func_name,
)
# L'output dettagliato è già loggato da log_and_execute a livello INFO # L'output dettagliato è già loggato da log_and_execute a livello INFO
# --- Rilevamento Conflitti (RC=1 e messaggi specifici) --- # --- Rilevamento Conflitti (RC=1 e messaggi specifici) ---
elif pull_result.returncode == 1 and ( elif pull_result.returncode == 1 and (
"conflict" in combined_output_lower or "conflict" in combined_output_lower
"automatic merge failed" in combined_output_lower or or "automatic merge failed" in combined_output_lower
"fix conflicts and then commit the result" in combined_output_lower or "fix conflicts and then commit the result" in combined_output_lower
): ):
result_info['status'] = 'conflict' # Stato specifico per conflitti result_info["status"] = "conflict" # Stato specifico per conflitti
result_info['message'] = f"Pull from '{remote_name}' resulted in merge conflicts. Please resolve them manually in '{repo_path}' and commit." result_info["message"] = (
log_handler.log_error(f"Merge conflict detected during pull from '{remote_name}'.", func_name=func_name) f"Pull from '{remote_name}' resulted in merge conflicts. Please resolve them manually in '{repo_path}' and commit."
)
log_handler.log_error(
f"Merge conflict detected during pull from '{remote_name}'.",
func_name=func_name,
)
# Non impostiamo 'exception' qui, lo stato 'conflict' è l'informazione chiave # Non impostiamo 'exception' qui, lo stato 'conflict' è l'informazione chiave
else: else:
# Altro Errore (RC != 0 e non è conflitto) # Altro Errore (RC != 0 e non è conflitto)
result_info['status'] = 'error' result_info["status"] = "error"
stderr_lower = stderr_full.lower() # Usa solo stderr per errori specifici stderr_lower = (
log_handler.log_error(f"Pull command failed for '{remote_name}' (RC={pull_result.returncode}). Stderr: {stderr_lower}", func_name=func_name) stderr_full.lower()
) # Usa solo stderr per errori specifici
log_handler.log_error(
f"Pull command failed for '{remote_name}' (RC={pull_result.returncode}). Stderr: {stderr_lower}",
func_name=func_name,
)
# Controlla errori specifici noti (simili a fetch) # Controlla errori specifici noti (simili a fetch)
auth_errors = ["authentication failed", "permission denied", "could not read username", "fatal: could not read password"] auth_errors = [
connection_errors = ["repository not found", "could not resolve host", "name or service not known", "network is unreachable"] "authentication failed",
upstream_errors = ["no tracking information", "no upstream branch", "refusing to merge unrelated histories"] # Errori legati a config/history "permission denied",
"could not read username",
"fatal: could not read password",
]
connection_errors = [
"repository not found",
"could not resolve host",
"name or service not known",
"network is unreachable",
]
upstream_errors = [
"no tracking information",
"no upstream branch",
"refusing to merge unrelated histories",
] # Errori legati a config/history
if any(err in stderr_lower for err in auth_errors): if any(err in stderr_lower for err in auth_errors):
result_info['message'] = f"Authentication required or failed for remote '{remote_name}' during pull." result_info["message"] = (
result_info['exception'] = GitCommandError(result_info['message'], stderr=pull_result.stderr) f"Authentication required or failed for remote '{remote_name}' during pull."
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=pull_result.stderr
)
elif any(err in stderr_lower for err in connection_errors): elif any(err in stderr_lower for err in connection_errors):
result_info['message'] = f"Failed to connect to remote '{remote_name}' during pull: Repository or host not found/reachable." result_info["message"] = (
result_info['exception'] = GitCommandError(result_info['message'], stderr=pull_result.stderr) f"Failed to connect to remote '{remote_name}' during pull: Repository or host not found/reachable."
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=pull_result.stderr
)
elif any(err in stderr_lower for err in upstream_errors): elif any(err in stderr_lower for err in upstream_errors):
result_info['message'] = f"Pull failed for '{remote_name}': Check branch upstream configuration or related history. Error: {pull_result.stderr.strip()}" result_info["message"] = (
result_info['exception'] = GitCommandError(result_info['message'], stderr=pull_result.stderr) f"Pull failed for '{remote_name}': Check branch upstream configuration or related history. Error: {pull_result.stderr.strip()}"
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=pull_result.stderr
)
else: else:
# Errore generico di Git # Errore generico di Git
result_info['message'] = f"Pull from '{remote_name}' failed (RC={pull_result.returncode}). Check logs." result_info["message"] = (
result_info['exception'] = GitCommandError(result_info['message'], stderr=pull_result.stderr) f"Pull from '{remote_name}' failed (RC={pull_result.returncode}). Check logs."
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=pull_result.stderr
)
except (GitCommandError, ValueError) as e: except (GitCommandError, ValueError) as e:
# Cattura errori dalla validazione iniziale o dal check dello stato # Cattura errori dalla validazione iniziale o dal check dello stato
log_handler.log_error(f"Error during pull execution for '{remote_name}': {e}", func_name=func_name) log_handler.log_error(
result_info = {'status': 'error', 'message': f"Pull failed: {e}", 'exception': e} f"Error during pull execution for '{remote_name}': {e}",
func_name=func_name,
)
result_info = {
"status": "error",
"message": f"Pull failed: {e}",
"exception": e,
}
except Exception as e: except Exception as e:
# Cattura errori imprevisti # Cattura errori imprevisti
log_handler.log_exception(f"Unexpected error during pull for '{remote_name}': {e}", func_name=func_name) log_handler.log_exception(
result_info = {'status': 'error', 'message': f"Unexpected pull error: {type(e).__name__}", 'exception': e} f"Unexpected error during pull for '{remote_name}': {e}",
func_name=func_name,
)
result_info = {
"status": "error",
"message": f"Unexpected pull error: {type(e).__name__}",
"exception": e,
}
return result_info return result_info
def execute_remote_push(self, repo_path: str, remote_name: str, branch_name: str): def execute_remote_push(
# To be implemented: Calls git_commands.git_push, handles upstream/errors self,
pass repo_path: str,
remote_name: str,
current_branch_name: str,
force: bool = False, # Aggiunto parametro force
) -> dict:
"""
Executes 'git push' for the current local branch to the specified remote.
Handles setting upstream for the first push and detects common errors.
def execute_push_tags(self, repo_path: str, remote_name: str): Args:
# To be implemented: Calls git_commands.git_push_tags repo_path (str): Path to the local repository.
pass remote_name (str): The name of the remote to push to (e.g., 'origin').
current_branch_name (str): The name of the currently checked-out local branch.
force (bool): Whether to force the push (use with caution!).
Returns:
dict: A dictionary containing the status ('success', 'rejected', 'error'),
a message, and optionally an exception.
"""
func_name = "execute_remote_push"
action_desc = f"Push branch '{current_branch_name}' to remote '{remote_name}'"
if force:
action_desc += " (FORCE PUSH)"
log_handler.log_info(
f"Executing: {action_desc} in '{repo_path}'", func_name=func_name
)
# --- Input Validation ---
if not repo_path or not os.path.isdir(repo_path):
raise ValueError(f"Invalid repository path: '{repo_path}'")
if not remote_name or remote_name.isspace():
raise ValueError("Remote name cannot be empty.")
if not current_branch_name or current_branch_name.isspace():
raise ValueError("Current branch name cannot be empty.")
if not os.path.exists(os.path.join(repo_path, ".git")):
raise ValueError(f"Directory '{repo_path}' is not a Git repository.")
# --- Pre-Push Checks ---
# 1. Verifica se siamo in detached HEAD (non dovremmo pushare da qui)
# (get_current_branch_name in git_commands già gestisce questo implicitamente,
# ma una verifica esplicita qui potrebbe dare un errore più chiaro)
# -> Omesso per ora, ma si potrebbe aggiungere se get_current_branch_name restituisce None.
# 2. Determina se l'upstream è già configurato
needs_set_upstream = False
try:
upstream = self.git_commands.get_branch_upstream(
repo_path, current_branch_name
)
if upstream is None:
needs_set_upstream = True
log_handler.log_info(
f"Upstream not set for branch '{current_branch_name}'. Will attempt push with --set-upstream.",
func_name=func_name,
)
else:
log_handler.log_debug(
f"Branch '{current_branch_name}' is tracking '{upstream}'.",
func_name=func_name,
)
except GitCommandError as e:
# Se il controllo upstream fallisce, non procedere
msg = f"Push aborted: Failed to check upstream configuration for branch '{current_branch_name}': {e}"
log_handler.log_error(msg, func_name=func_name)
return {"status": "error", "message": msg, "exception": e}
# --- Esecuzione Git Push ---
result_info = {"status": "unknown", "message": "Push not completed."}
try:
# Chiama git_push passando il flag set_upstream e force
push_result = self.git_commands.git_push(
working_directory=repo_path,
remote_name=remote_name,
branch_name=current_branch_name,
set_upstream=needs_set_upstream,
force=force,
)
# Analizza il risultato del comando push
stdout_full = push_result.stdout if push_result.stdout else ""
stderr_full = push_result.stderr if push_result.stderr else ""
combined_output_lower = (stdout_full + stderr_full).lower()
if push_result.returncode == 0:
# Successo
result_info["status"] = "success"
if "everything up-to-date" in combined_output_lower:
result_info["message"] = (
f"Push to '{remote_name}': Branch '{current_branch_name}' already up-to-date."
)
log_handler.log_info(
f"Push successful (already up-to-date) for '{current_branch_name}'.",
func_name=func_name,
)
else:
result_info["message"] = (
f"Push branch '{current_branch_name}' to '{remote_name}' completed successfully."
)
log_handler.log_info(
f"Push successful for '{current_branch_name}'. Output logged.",
func_name=func_name,
)
# L'output dettagliato è già loggato da log_and_execute a livello INFO
# --- Rilevamento Push Rifiutato (RC=1 e messaggi specifici) ---
elif push_result.returncode == 1 and (
"rejected" in stderr_full.lower()
or "failed to push some refs" in stderr_full.lower()
):
# Controlla se il rifiuto è dovuto a divergenza (non-fast-forward)
is_non_fast_forward = (
"non-fast-forward" in stderr_full.lower()
or "updates were rejected because the remote contains work"
in stderr_full.lower()
)
if is_non_fast_forward and not force:
result_info["status"] = "rejected" # Stato specifico per rifiuto
result_info["message"] = (
f"Push rejected: Remote repository has changes you don't have locally for branch '{current_branch_name}'.\nHint: Try pulling first to integrate remote changes."
)
log_handler.log_error(
f"Push rejected (non-fast-forward) for '{current_branch_name}'. User needs to pull.",
func_name=func_name,
)
elif force and is_non_fast_forward:
# Se era un force push e viene rifiutato comunque, è un errore diverso
result_info["status"] = "error"
result_info["message"] = (
f"FORCE PUSH rejected for '{current_branch_name}'. Reason: {stderr_full.strip()}"
)
log_handler.log_error(
f"FORCE PUSH rejected for '{current_branch_name}'. Stderr: {stderr_full.strip()}",
func_name=func_name,
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=push_result.stderr
)
else:
# Altro tipo di rifiuto (es. pre-receive hook)
result_info["status"] = "rejected"
result_info["message"] = (
f"Push rejected for '{current_branch_name}'. Reason: {stderr_full.strip()}"
)
log_handler.log_error(
f"Push rejected for '{current_branch_name}'. Stderr: {stderr_full.strip()}",
func_name=func_name,
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=push_result.stderr
)
else:
# Altro Errore (RC != 0 e non è rifiuto specifico)
result_info["status"] = "error"
stderr_lower = (
stderr_full.lower()
) # Usa solo stderr per errori specifici
log_handler.log_error(
f"Push command failed for '{current_branch_name}' (RC={push_result.returncode}). Stderr: {stderr_lower}",
func_name=func_name,
)
# Controlla errori specifici noti (auth, connessione)
auth_errors = [
"authentication failed",
"permission denied",
"could not read username",
"fatal: could not read password",
]
connection_errors = [
"repository not found",
"could not resolve host",
"name or service not known",
"network is unreachable",
]
if any(err in stderr_lower for err in auth_errors):
result_info["message"] = (
f"Authentication required or failed for remote '{remote_name}' during push."
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=push_result.stderr
)
elif any(err in stderr_lower for err in connection_errors):
result_info["message"] = (
f"Failed to connect to remote '{remote_name}' during push: Repository or host not found/reachable."
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=push_result.stderr
)
# Potrebbe esserci l'errore "src refspec <branch> does not match any" se il branch locale non esiste
elif (
f"src refspec {current_branch_name} does not match any"
in stderr_lower
):
result_info["message"] = (
f"Push failed: Local branch '{current_branch_name}' not found."
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=push_result.stderr
)
else:
# Errore generico di Git
result_info["message"] = (
f"Push to '{remote_name}' failed (RC={push_result.returncode}). Check logs."
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=push_result.stderr
)
except (GitCommandError, ValueError) as e:
# Cattura errori dalla validazione o dal check upstream
log_handler.log_error(
f"Error during push execution for '{current_branch_name}': {e}",
func_name=func_name,
)
result_info = {
"status": "error",
"message": f"Push failed: {e}",
"exception": e,
}
except Exception as e:
# Cattura errori imprevisti
log_handler.log_exception(
f"Unexpected error during push for '{current_branch_name}': {e}",
func_name=func_name,
)
result_info = {
"status": "error",
"message": f"Unexpected push error: {type(e).__name__}",
"exception": e,
}
return result_info
def execute_push_tags(self, repo_path: str, remote_name: str) -> dict:
"""
Executes 'git push --tags' to the specified remote.
Args:
repo_path (str): Path to the local repository.
remote_name (str): The name of the remote to push tags to.
Returns:
dict: A dictionary containing the status ('success', 'error'), message,
and optionally an exception.
"""
func_name = "execute_push_tags"
log_handler.log_info(
f"Executing push tags to remote '{remote_name}' in '{repo_path}'",
func_name=func_name,
)
# --- Input Validation ---
if not repo_path or not os.path.isdir(repo_path):
raise ValueError(f"Invalid repository path: '{repo_path}'")
if not remote_name or remote_name.isspace():
raise ValueError("Remote name cannot be empty.")
if not os.path.exists(os.path.join(repo_path, ".git")):
raise ValueError(f"Directory '{repo_path}' is not a Git repository.")
# --- Esecuzione Git Push Tags ---
result_info = {"status": "unknown", "message": "Push tags not completed."}
try:
# Chiama git_push_tags (che ha check=False)
push_result = self.git_commands.git_push_tags(repo_path, remote_name)
# Analizza il risultato
stderr_full = push_result.stderr if push_result.stderr else ""
stderr_lower = stderr_full.lower()
if push_result.returncode == 0:
# Successo
result_info["status"] = "success"
# Verifica se c'era qualcosa da pushare
if (
"everything up-to-date" in (push_result.stdout or "").lower()
or "everything up-to-date" in stderr_lower
): # A volte l'output va su stderr
result_info["message"] = (
f"Push tags to '{remote_name}': All tags already up-to-date."
)
log_handler.log_info(
f"Push tags successful (already up-to-date) for '{remote_name}'.",
func_name=func_name,
)
else:
result_info["message"] = (
f"Push tags to '{remote_name}' completed successfully."
)
log_handler.log_info(
f"Push tags successful for '{remote_name}'. Output logged.",
func_name=func_name,
)
# Non c'è un caso 'rejected' specifico per i tag come per i branch (non-fast-forward non applicabile)
# Quindi trattiamo RC != 0 come errore generico.
else:
result_info["status"] = "error"
log_handler.log_error(
f"Push tags command failed for '{remote_name}' (RC={push_result.returncode}). Stderr: {stderr_lower}",
func_name=func_name,
)
# Controlla errori specifici (auth, connessione)
auth_errors = [
"authentication failed",
"permission denied",
"could not read username",
"fatal: could not read password",
]
connection_errors = [
"repository not found",
"could not resolve host",
"name or service not known",
"network is unreachable",
]
if any(err in stderr_lower for err in auth_errors):
result_info["message"] = (
f"Authentication required or failed for remote '{remote_name}' during push tags."
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=push_result.stderr
)
elif any(err in stderr_lower for err in connection_errors):
result_info["message"] = (
f"Failed to connect to remote '{remote_name}' during push tags."
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=push_result.stderr
)
else:
result_info["message"] = (
f"Push tags to '{remote_name}' failed (RC={push_result.returncode}). Check logs."
)
result_info["exception"] = GitCommandError(
result_info["message"], stderr=push_result.stderr
)
except (GitCommandError, ValueError) as e:
log_handler.log_error(
f"Error during push tags execution for '{remote_name}': {e}",
func_name=func_name,
)
result_info = {
"status": "error",
"message": f"Push tags failed: {e}",
"exception": e,
}
except Exception as e:
log_handler.log_exception(
f"Unexpected error during push tags for '{remote_name}': {e}",
func_name=func_name,
)
result_info = {
"status": "error",
"message": f"Unexpected push tags error: {type(e).__name__}",
"exception": e,
}
return result_info
# --- END OF FILE remote_actions.py --- # --- END OF FILE remote_actions.py ---