[foreign_testing] New project to support foreign testing in FORGE

projects/core/ci_entrypoint/github/pr_args.py

@@ -69,10 +69,10 @@ def handle_test_directive(line: str) -> Dict[str, Any]:
 
     test_name = parts.pop(0)
 
-    if not parts:
-        raise ValueError(f"Found a /test directive without a project ('{line.strip()}')")
-
-    project_name = parts.pop(0)
+    if parts:
+        project_name = parts.pop(0)
+    else:
+        project_name = "project_not_set"
 
     args = parts # allowed to be empty
     result = {}

projects/core/ci_entrypoint/prepare_ci.py

@@ -556,9 +556,19 @@ def send_notification(project: str, operation: str, finish_reason: FinishReason,
         success = finish_reason == FinishReason.SUCCESS
         notification_status = f"Test of '{project} {operation}' {('succeeded' if success else 'failed')}{duration}"
 
-        # Skip notifications for successful non-test steps
-        if success and operation not in ("test", "submit"):
-            logger.info(f"Skipping notification for successful '{operation}' step (only 'test' and 'submit' steps notify on success)")
+        if project == "foreign_testing":
+            # never submit for the time being
+            logger.info(f"Skipping notification for project '{project}'")
+            return
+        elif project == "fournos_launcher":
+            # always submit for the time being
+            logger.info(f"Keeping all the notications for the {project} project")
+            pass
+        elif os.environ.get("PSAP_FORGE_FOREIGN_TESTING"):
+            logger.info("Keeping all the notications for foreign tests")
+            pass
+        elif success and operation not in ("test", ):
+            logger.info(f"Skipping notification for successful '{operation}' step (only 'test' step notifies on success)")
             return
 
         # Enable GitHub notifications by default, Slack can be enabled via environment variable

projects/core/ci_entrypoint/run_ci.py

@@ -17,6 +17,8 @@
     run skeleton validate
 """
 
+# Import CI preparation module
+from projects.core.ci_entrypoint import prepare_ci
 
 import os
 import sys
@@ -27,20 +29,33 @@
 from pathlib import Path
 from typing import List, Optional
 
+import click
+
+def setup_logging():
+    """Set up logging configuration."""
+    logging.basicConfig(
+        level=logging.INFO,
+        format='%(levelname)s: %(message)s',
+        handlers=[logging.StreamHandler(sys.stderr)]
+    )
+
+# Set up logging
+setup_logging()
+logger = logging.getLogger(__name__)
+
 FORGE_HOME = Path(__file__).resolve().parent.parent.parent.parent
 
-EXTRA_PACKAGES = ["fire"]
+EXTRA_PACKAGES = []
 
 def signal_handler_sigint(sig, frame):
     """Handle SIGINT (Ctrl+C) gracefully."""
     print(f"\n🚫 Received SIGINT (Ctrl+C) - Interrupting CI operation...")
 
     # Emergency cleanup of dual output
-    if prepare_ci:
-        try:
-            prepare_ci.shutdown_dual_output()
-        except Exception:
-            pass  # Don't let cleanup errors prevent signal handling
+    try:
+        prepare_ci.shutdown_dual_output()
+    except Exception:
+        pass  # Don't let cleanup errors prevent signal handling
 
     sys.exit(130)  # Standard exit code for SIGINT
 
@@ -50,11 +65,10 @@ def signal_handler_sigterm(sig, frame):
     print(f"\n🛑 Received SIGTERM - Terminating CI operation...")
 
     # Emergency cleanup of dual output
-    if prepare_ci:
-        try:
-            prepare_ci.shutdown_dual_output()
-        except Exception:
-            pass  # Don't let cleanup errors prevent signal handling
+    try:
+        prepare_ci.shutdown_dual_output()
+    except Exception:
+        pass  # Don't let cleanup errors prevent signal handling
 
     sys.exit(143)  # Standard exit code for SIGTERM
 
@@ -72,15 +86,9 @@ def setup_signal_handlers():
         pass
 
 
-def setup_logging():
-    """Set up logging configuration."""
-    logging.basicConfig(
-        level=logging.INFO,
-        format='%(levelname)s: %(message)s',
-        handlers=[logging.StreamHandler(sys.stderr)]
-    )
-
 def install_extra_packages(packages):
+    if not packages:
+        return
 
     print(f"📦 Installing {'/'.join(packages)} packages...")
 
@@ -130,32 +138,23 @@ def install_extra_packages(packages):
     import importlib
     importlib.invalidate_caches()
 
-# Set up logging
-setup_logging()
-logger = logging.getLogger(__name__)
-
-# Set up signal handlers for graceful interruption
-setup_signal_handlers()
 
-# Install click package using uv (as non-root user)
-install_extra_packages(EXTRA_PACKAGES)
+def prepare():
+    # Set up signal handlers for graceful interruption
+    setup_signal_handlers()
 
-import click
-
-# Import CI preparation module
-try:
-    import prepare_ci
-    logger.info("CI preparation module imported successfully")
+    # Install click package using uv (as non-root user)
+    install_extra_packages(EXTRA_PACKAGES)
 
     # Set up ARTIFACT_DIR
     prepare_ci.precheck_artifact_dir()
 
-    # Set up dual output as early as possible
-    prepare_ci.setup_dual_output()
-
-except ImportError as e:
-    logger.warning(f"CI preparation module not available: {e}")
-    prepare_ci = None
+    if os.isatty(sys.stdin.fileno()):
+        # not working very well from a TTY ...
+        logger.info("Running from a TTY, not enabling the dual output")
+    else:
+        # Set up dual output as early as possible
+        prepare_ci.setup_dual_output()
 
 
 def find_project_directory(project_name: str) -> Optional[Path]:
@@ -202,6 +201,30 @@ def find_ci_script(project_dir: Path, operation: str) -> Optional[Path]:
     return None
 
 
+def find_ci_script(project_dir: Path, operation: str) -> Optional[Path]:
+    """
+    Find the appropriate CI script for the operation.
+
+    Args:
+        project_dir: Project directory path
+        operation: Operation to perform (e.g., 'ci')
+
+    Returns:
+        Path to CI script if found, None otherwise
+    """
+    # Check possible locations for CI scripts
+    possible_locations = [
+        # Operation-specific script in orchestration subdirectory
+        project_dir / "orchestration" / f"{operation}.py",
+    ]
+
+    for script_path in possible_locations:
+        if script_path.exists() and os.access(script_path, os.X_OK):
+            return script_path
+
+    return None
+
+
 def get_available_projects() -> List[str]:
     """Get list of available projects."""
 
@@ -337,18 +360,30 @@ def parse_cli_help(help_output: str) -> List[str]:
     return operations
 
 
-def execute_project_operation(project: str, operation: str, args: tuple, verbose: bool, dry_run: bool):
+def execute_project_operation(
+        project: str,
+        operation: str,
+        args: tuple,
+        verbose: bool = False,
+        dry_run: bool  = False,
+        do_prepare_ci: bool = True
+):
     """Execute a project operation."""
+
+    if not (isinstance(args, list) or isinstance(args, tuple)):
+        raise ValueError(f"Args={args} must be a list, not {args.__class__.__name__}")
+
+
     if verbose:
         click.echo("")
         click.echo(f"🚀 FORGE CI Orchestration")
         click.echo(f"Project: {project}")
         click.echo(f"Operation: {operation}")
-        click.echo(f"Arguments: {list(args)}")
+        click.echo(f"Arguments: {' '.join(args)}")
         click.echo("")
 
     # Execute CI preparation tasks
-    if prepare_ci:
+    if do_prepare_ci:
         try:
             prepare_ci.prepare(
                 verbose=verbose,
@@ -363,7 +398,7 @@ def execute_project_operation(project: str, operation: str, args: tuple, verbose
             )
             raise
     else:
-        logger.warning("CI preparation module not available, skipping preparation")
+        logger.warning("CI preparation not enabled, skipping preparation")
 
     # Find project directory
     project_dir = find_project_directory(project)
@@ -386,14 +421,26 @@ def execute_project_operation(project: str, operation: str, args: tuple, verbose
     # Find CI script
     ci_script = find_ci_script(project_dir, operation)
     if not ci_script:
-        click.echo(
-            click.style(
-                f"❌ ERROR: No CI script found for project '{project}' operation '{operation}'.",
-                fg='red'
-            ),
-            err=True
-        )
-        click.echo(f"🔍 Expected: {project_dir}/orchestration/{operation}.py")
+        script_path = project_dir / "orchestration" / f"{operation}.py"
+
+        if script_path.exists():
+            click.echo(
+                click.style(
+                    f"❌ ERROR: CI script exists but is not executable for project '{project}' operation '{operation}'.",
+                    fg='red'
+                ),
+                err=True
+            )
+            click.echo(f"💡 Fix with: chmod +x {script_path}")
+        else:
+            click.echo(
+                click.style(
+                    f"❌ ERROR: No CI script found for project '{project}' operation '{operation}'.",
+                    fg='red'
+                ),
+                err=True
+            )
+            click.echo(f"🔍 Expected: {script_path}")
         sys.exit(1)
 
     # Convert underscores to hyphens in args for Click compatibility
@@ -441,35 +488,26 @@ def execute_project_operation(project: str, operation: str, args: tuple, verbose
 
         success = finish_reason == prepare_ci.FinishReason.SUCCESS
         # Post-execution checks and status reporting
-        if prepare_ci:
-            status_message = prepare_ci.postchecks(project, operation, start_time, finish_reason, list(args))
-            msg = click.style(status_message, \
-                              fg='green' if success else 'red')
-        else:
-            # Fallback to simple messages if prepare_ci not available
-            if success:
-                msg = click.style(f"✅ {project} {operation} completed successfully", fg='green')
-            else:
-                msg = click.style(f"❌ {project} {operation} failed with exit code {result.returncode}", fg='red')
+        status_message = prepare_ci.postchecks(project, operation, start_time, finish_reason, list(args))
+        msg = click.style(status_message, \
+                          fg='green' if success else 'red')
 
         click.echo()
         click.echo(msg, err=not success)
 
-        if prepare_ci:
-            # Properly shutdown dual output to flush all buffers and terminate daemon
-            prepare_ci.shutdown_dual_output()
+        # Properly shutdown dual output to flush all buffers and terminate daemon
+        prepare_ci.shutdown_dual_output()
 
         sys.exit(result.returncode)
 
     except Exception as e:
         logging.exception("Unexpected exception")
 
         # Emergency cleanup of dual output to prevent hanging
-        if prepare_ci:
-            try:
-                prepare_ci.shutdown_dual_output()
-            except Exception:
-                pass  # Don't let cleanup errors mask the original error
+        try:
+            prepare_ci.shutdown_dual_output()
+        except Exception:
+            pass  # Don't let cleanup errors mask the original error
 
         click.echo(
             click.style(f"❌ ERROR: Unexpected error during execution", fg='red'),
@@ -501,9 +539,7 @@ def main(project, operation, args, verbose, dry_run):
         run skeleton validate
     """
 
-    if project == "fournos":
-        logger.info("Using project 'fournos_launcher' instead of 'fournos'")
-        project = "fournos_launcher"
+    prepare()
 
     # No arguments - list projects
     if not project:

projects/core/library/config.py

@@ -55,6 +55,12 @@ def __init__(self, config_path):
         with open(self.config_path) as config_f:
             self.config = yaml.safe_load(config_f)
 
+        if self.config is None:
+            self.config = {}
+
+        if not isinstance(self.config, dict):
+            raise ValueError(f"YAML loaded from {self.config_path} isn't a dictionnary ({self.config.__class__.__name__})")
+
 
     def ensure_core_fields(self):
         """

projects/core/notifications/github/api.py

@@ -24,6 +24,9 @@ def get_user_token(pem_file, client_id, org, repo):
 
     installation_resp = requests.get(f"{BASE_URL}/repos/{org}/{repo}/installation", headers=headers)
 
+    if "id" not in installation_resp.json():
+        return None
+
     try:
         installation_id = installation_resp.json()["id"]
     except Exception as e:

projects/core/notifications/send.py

@@ -92,6 +92,10 @@ def send_job_completion_notification_to_github(pem_file, client_id, finish_reaso
         return
 
     user_token = github_api.get_user_token(pem_file, client_id, org, repo)
+    if not user_token:
+        logging.error("github: Couldn't fetch the user token. Is the app installed in the repo?")
+        return
+
     if dry_run:
         logging.info(f"Github notification:\n{message}")
         logging.info(f"***")