9pm.py

#!/usr/bin/env python3

import argparse
import errno
import json
import os
import pty
import yaml
import subprocess
import sys
import time
import tempfile
import shutil
import re
import atexit
import hashlib
import shlex
import textwrap
from collections import deque
from datetime import datetime

TEST_CNT = 0
SUITE_CNT = 0
ROOT_PATH = os.path.dirname(os.path.realpath(__file__))
# TODO: proper argument strucutre
DATABASE = ""
SCRATCHDIR = ""
LOGDIR = None
VERBOSE = False
NOEXEC = False
PROJECT = {}

class pcolor:
    purple = '\033[95m'
    blue = '\033[94m'
    green = '\033[92m'
    yellow = '\033[93m'
    yellow_u = '\033[4;33m'
    red = '\033[91m'
    red_u ='\033[4;31m'
    cyan = '\033[96m'
    reset = '\033[0m'
    orange = '\033[33m'
    faint = '\033[2m'

def cprint(color, *args, **kwargs):
    end = kwargs.pop('end', '\n')
    sys.stdout.write(color)
    print(*args, **kwargs, end='')
    sys.stdout.write(pcolor.reset + end)

def vcprint(color, *args, **kwargs):
    global VERBOSE

    if VERBOSE:
        cprint(color, *args, **kwargs)

def rootify_path(path):
    path = os.path.join(ROOT_PATH, path)
    path = os.path.expanduser(path)
    path = os.path.normpath(path)
    return path

def proj_relpath(path):
    """Path relative to the project root, or unchanged if outside it"""
    root = os.path.realpath(PROJECT['PROJECT-ROOT'])
    rel = os.path.relpath(os.path.realpath(path), root)
    return os.path.normpath(path) if rel.startswith(os.pardir) else rel

def pty_read_lines(fd):
    # PTY line discipline turns '\n' into '\r\n' on the way out, so trailing
    # '\r' is stripped from each yielded line.
    buf = b""
    while True:
        try:
            chunk = os.read(fd, 4096)
        except OSError as e:
            if e.errno != errno.EIO:
                raise
            chunk = b""
        if not chunk:
            if buf:
                yield buf.rstrip(b'\r')
            return
        buf += chunk
        while b'\n' in buf:
            line, buf = buf.split(b'\n', 1)
            yield line.rstrip(b'\r')

def execute(args, test, output_log):
    os.environ["NINEPM_TEST_NAME"] = test['unix_name']

    # Run the test with stdout connected to a pty so its stdio sees a TTY and
    # line-buffers, instead of block-buffering through a pipe until exit.
    master_fd, slave_fd = pty.openpty()
    proc = subprocess.Popen([test['case']] + args, stdout=slave_fd)
    os.close(slave_fd)

    skip_suite = False
    test_skip = False
    err = False
    test['failures'] = []
    context = deque(maxlen=10)  # output since the last TAP result line
    failure = None              # lines of the failure being collected

    # Test metadata is now handled in the report generation, not in the log

    for raw in pty_read_lines(master_fd):
        string = raw.decode('utf-8', errors='replace').rstrip()
        stamp = time.strftime("%Y-%m-%d %H:%M:%S")

        plan = re.search(r'^(\d+)..(\d+)$', string)
        ok = re.search(r'^ok (\d+) -', string)
        not_ok = re.search(r'^not ok (\d+) -', string)
        skip = re.search(r'^ok (\d+) # skip', string)
        skip_suite = re.search(r'^ok (\d+) # skip suite', string)
        comment = re.search(r'^\w*#', string)

        output_log.write(f"{stamp} {string}\n")

        # Collect what the test said around a failure: the output leading up
        # to the not ok line and what follows it, until the next TAP line.
        if plan or ok or not_ok or skip:
            failure = None
            if not_ok:
                failure = list(context) + [string]
                test['failures'].append(failure)
            context.clear()
        else:
            context.append(string)
            if failure is not None and len(failure) < 40:
                failure.append(string)

        if plan:
            cprint(pcolor.purple, f'{stamp} {string}')
            test['plan'] = plan.group(2)
        elif skip:
            cprint(pcolor.yellow, f'{stamp} {string}')
            test['executed'] = skip.group(1)
            test_skip = True
        elif skip_suite:
            cprint(pcolor.yellow, f'{stamp} {string}')
            test['executed'] = skip.group(1)
            skip_suite = True
            test_skip = True
        elif ok:
            cprint(pcolor.green, f'{stamp} {string}')
            test['executed'] = ok.group(1)
        elif not_ok:
            cprint(pcolor.red, f'{stamp} {string}')
            err = True
            test['executed'] = not_ok.group(1)
        elif comment:
            cprint(pcolor.faint, f'{stamp} {string}')
        else:
            print(f"{stamp} {string}")

    os.close(master_fd)
    proc.wait()
    exitcode = proc.returncode

    if exitcode != 0:
        err = True

    return skip_suite, test_skip, err

def calculate_sha1sum(path):
    sha1 = hashlib.sha1()
    with open(path, "rb") as file:
        for chunk in iter(lambda: file.read(4096), b""):
            sha1.update(chunk)
    return sha1.hexdigest()

def run_onfail(args, test):
    opts = []
    if 'options' in test:
        opts.extend(test['options'])

    dirname = os.path.dirname(test['case'])

    onfail = {}
    onfail['case'] = os.path.join(dirname, test['onfail'])
    onfail['unix_name'] = 'onfail'
    onfail['name'] = 'onfail'

    print(f"\n{pcolor.cyan}Running onfail \"{test['onfail']}\" for test {test['name']}{pcolor.reset}")

    vcprint(pcolor.faint, f"Executing onfail {onfail['case']} for test {test['case']}")

    with open(os.path.join(LOGDIR, "on-fail.log"), 'a') as log:
        log.write(f"\n\nON FAIL START")
        if not NOEXEC:
            execute(opts, onfail, log)

def run_test(args, test):
    opts = []
    if 'options' in test:
        opts.extend(test['options'])

    name = test['name']
    path = proj_relpath(test['case'])
    print(f"\n{pcolor.blue}Starting test {test['uniq_id']}: \"{test['uniq_id']:04d} {name}\" ({path}){pcolor.reset}")

    if test['result'] == "skip":
        print(f"{pcolor.yellow}Skip test {name} (suite skip){pcolor.reset}")
        # Delete outfile as this test won't have any output
        if 'outfile' in test:
            del test['outfile']

        return True, True, False

    vcprint(pcolor.faint, f"Test File: {test['case']}")
    vcprint(pcolor.faint, f"Test Cmdl: {opts}")

    with open(os.path.join(LOGDIR, test['outfile']), 'a') as output:
        if NOEXEC:
            print("Skipped because --no-exec", file=output)
            return False, True, False
        else:
            skip_suite, skip, err = execute(opts, test, output)

    if 'plan' not in test:
        print("test error, no plan")
        return False, False, True

    if 'executed' not in test:
        print("test error, no tests executed")
        return False, False, True

    if test['plan'] != test['executed']:
        print(f"test error, not conforming to plan ({test['executed']}/{test['plan']})")
        err = True

    return skip_suite, skip, err


def slugify(text, lowercase=True):
    """
    Local good-enough slugify() replacement

    Replace spaces, special chars, and non-ASCII chars with '-',
    then squash any resulting repeated '-' for readability.
    """
    if lowercase:
        text = text.lower()

    s = re.sub(r'[^\w-]', '-', text, flags=re.ASCII)
    return re.sub(r'-{2,}', '-', s)


# In this function, we generate an unique name for each case and suite. Both
# suites and cases can be passed an arbitrary amount of times and the same test
# can reside in different suites. We need something unique to identify them by.
def prefix_name(name):
    global TEST_CNT
    TEST_CNT += 1
    return TEST_CNT, slugify(name, lowercase=True)

def next_suite_num():
    global SUITE_CNT
    SUITE_CNT += 1
    return SUITE_CNT

def gen_name(filepath):
    return os.path.basename(filepath)

def gen_unix_name(filename):
    base = os.path.basename(filename)
    name_without_ext = os.path.splitext(base)[0]
    return prefix_name(name_without_ext)

def gen_outfile(name):
    return os.path.join("output", os.path.splitext(name)[0] + ".log")

def lmerge(a, b):
    new = a.copy()
    for item in b:
        if item not in a:
            new.append(item)
    return new

def parse_yaml(path):
    with open(path, 'r') as stream:
        try:
            data = yaml.load(stream, Loader=yaml.FullLoader)
        except yaml.YAMLError as exc:
            print(exc)
            return -1
    return data

def get_test_spec_path(case_path, test_spec):
    case_dirname = os.path.dirname(case_path)
    case_basename = os.path.basename(case_path)
    case_name = os.path.splitext(case_basename)[0]

    test_spec = test_spec.replace('<case>', case_name)

    return  os.path.join(case_dirname, test_spec)


def get_suite_settings(name, data, upstream_settings):
    for entry in data:
        if 'settings' in entry:
            if entry['settings'] is None:
                print(f"error, empty \"settings\" in suite {suite['suite_path']}, invalid indent?")
                sys.exit(1)

            vcprint(pcolor.faint, f"Suite {name} has settings: {entry['settings']}")
            return entry['settings']
    return upstream_settings

def parse_suite(suite_path, parent_suite_path, options, settings, name=None):
    suite = {}
    suite['suite'] = []
    suite['result'] = "pending"
    suite_dirname = os.path.dirname(suite_path)

    suite['name'] = name if name else gen_name(suite_path)
    suite['suite_num'] = next_suite_num()
    suite['suite_path'] = suite_path

    if not os.path.isfile(suite_path):
        print(f"error, test suite not found {suite_path}")
        print(f"(referenced from {parent_suite_path})")
        sys.exit(1)

    data = parse_yaml(suite_path)
    if not data:
        print(f"fatal, empty suite {suite['name']}")
        sys.exit(1)

    settings = get_suite_settings(suite['name'], data, settings)

    for entry in data:
        if 'suite' in entry:
            next_suite_path = os.path.join(suite_dirname, entry['suite'])
            if 'opts' in entry:
                opts = [o.replace('<base>', suite_dirname) for o in entry['opts']]
                opts = [o.replace('<scratch>', SCRATCHDIR) for o in opts]
                opts = lmerge(opts, options)
            else:
                opts = options.copy()
            sub = parse_suite(next_suite_path, suite_path, opts, settings, entry.get('name'))
            if 'abort' in entry:
                sub['abort'] = entry['abort']
            suite['suite'].append(sub)

        elif 'case' in entry:
            case = {}

            if 'name' in entry:
                uniq_id, uname = prefix_name(entry['name'])
                case['uniq_id'] = uniq_id
                case['unix_name'] = f"{uniq_id:04d}-{uname}"
                case['name'] = entry['name']
            else:
                uniq_id, uname = gen_unix_name(entry['case'])
                case['uniq_id'] = uniq_id
                case['unix_name'] = f"{uniq_id:04d}-{uname}"
                case['name'] = gen_name(entry['case'])

            case['outfile'] = gen_outfile(case['unix_name'])

            if 'opts' in entry:
                opts = [o.replace('<base>', suite_dirname) for o in entry['opts']]
                opts = [o.replace('<scratch>', SCRATCHDIR) for o in opts]
                case['options'] = lmerge(opts, options)
            else:
                case['options'] = options

            if 'onfail' in entry:
                case['onfail'] = entry['onfail']

            if 'mask' in entry:
                case['mask'] = entry['mask']

            if 'abort' in entry:
                case['abort'] = entry['abort']

            case['case'] = os.path.join(suite_dirname, entry['case'])

            if 'test-spec' in settings:
                test_spec_path = get_test_spec_path(case['case'], settings['test-spec'])
                if os.path.exists(test_spec_path):
                    vcprint(pcolor.faint, f"Found test specification: {test_spec_path} for {case['case']}")
                    case['test_spec'] = test_spec_path
                    case['test_spec_sha'] = calculate_sha1sum(test_spec_path)
                else:
                        vcprint(pcolor.faint, f"No test specification for {case['case']} ({test_spec_path})")

            if not os.path.isfile(case['case']):
                print(f"error, test case not found {case['case']}")
                print(f"(referenced from {suite_path})")
                sys.exit(1)
            if not os.access(case['case'], os.X_OK):
                print(f"error, test case not executable {case['case']}")
                sys.exit(1)
            suite['suite'].append(case)
        elif 'settings' in entry:
            pass # Handled by preparser
        else:
            print(f"error, missing suite/case/settings in suite {suite['name']}")
            sys.exit(1)
    return suite

def collect_test_logs(test_data, depth=0):
    """Recursively collect all test logs and embed them in the data structure."""
    if not test_data.get('suite'):
        return

    for test in test_data['suite']:
        if 'outfile' in test:
            log_path = os.path.join(LOGDIR, test['outfile'])
            try:
                with open(log_path, 'r') as f:
                    test['logs'] = f.read()
            except FileNotFoundError:
                test['logs'] = f"Log file not found: {log_path}"
            except Exception as e:
                test['logs'] = f"Error reading log file {log_path}: {e}"

        if 'suite' in test:
            collect_test_logs(test, depth + 1)

def calculate_test_summary(test_data):
    """Calculate summary statistics for all tests."""
    counts = {
        'pass': 0,
        'fail': 0,
        'skip': 0,
        'masked_fail': 0,
        'masked_skip': 0,
        'total': 0
    }

    def count_tests(data):
        if not data.get('suite'):
            return

        for test in data['suite']:
            if 'suite' in test:
                # This is a sub-suite, recurse but don't count it
                count_tests(test)
            elif 'result' in test:
                # This is a leaf test case, count it
                result = test['result']
                if result == 'pass':
                    counts['pass'] += 1
                elif result == 'fail':
                    counts['fail'] += 1
                elif result == 'skip':
                    counts['skip'] += 1
                elif result == 'masked-fail':
                    counts['masked_fail'] += 1
                elif result == 'masked-skip':
                    counts['masked_skip'] += 1
                counts['total'] += 1

    count_tests(test_data)
    return counts

def write_json_result(data, agent_prompt=None):
    """Write comprehensive JSON result file with embedded logs."""
    # Collect all test logs and embed them in the data structure
    collect_test_logs(data)

    # Calculate summary statistics
    summary = calculate_test_summary(data)

    # Prepare metadata
    current_time = datetime.now()
    project_info = {
        'root': PROJECT['PROJECT-ROOT'],
        'version': PROJECT['PROJECT-VERSION'],
        'sha': PROJECT['PROJECT-SHA'],
    }
    if 'PROJECT-NAME' in PROJECT:
        project_info['name'] = PROJECT['PROJECT-NAME']
    if 'PROJECT-TOPDOC' in PROJECT:
        project_info['topdoc'] = PROJECT['PROJECT-TOPDOC']

    # Get 9pm version info
    ninepm_sha = run_git_cmd(ROOT_PATH, ['rev-parse', 'HEAD'])

    # Build complete JSON structure
    json_data = {
        'metadata': {
            'timestamp': current_time.isoformat(),
            'date': current_time.strftime("%Y-%m-%d"),
            'project': project_info,
            'environment': {
                '9pm_version': ninepm_sha[:10] if ninepm_sha else 'unknown',
                'log_dir': LOGDIR,
                'scratch_dir': SCRATCHDIR,
                'root_path': ROOT_PATH
            }
        },
        'summary': summary,
        'agent_prompt': agent_prompt,
        'suite': data
    }

    # Write JSON file
    json_path = os.path.join(LOGDIR, 'result.json')
    with open(json_path, 'w') as f:
        json.dump(json_data, f, indent=2, ensure_ascii=False)

    return json_path

def print_result_tree(data, base):
    i = 1
    llen = len(data['suite'])

    for test in data['suite']:
        if i < llen:
            prefix = "|-- "
            nextbase = base + "|   "
        else:
            prefix = "`-- "
            nextbase = base + "    "

        if test['result'] == "pass":
            sign = "o"
            color = pcolor.green
        elif test['result'] == "fail":
            sign = "x"
            color = pcolor.red
        elif test['result'] == "masked-fail":
            sign = "m"
            color = pcolor.red_u
        elif test['result'] == "masked-skip":
            sign = "m"
            color = pcolor.yellow_u
        elif test['result'] == "skip":
            sign = "s"
            color = pcolor.yellow
        else:
            sign = "?"
            color = pcolor.yellow

        if 'case' in test:
            print(f"{base}{prefix}{color}{sign} {test['uniq_id']:04d} {test['name']}{pcolor.reset}")
        else:
            print(f"{base}{prefix}{color}{sign} {test['name']}{pcolor.reset}")

        if 'suite' in test:
            print_result_tree(test, nextbase)
        i += 1

def collect_failures(data):
    failed = []
    for node in data['suite']:
        if 'suite' in node:
            failed.extend(collect_failures(node))
        elif node.get('result') == "fail" and node.get('failures'):
            failed.append(node)
    return failed

def project_label():
    name = PROJECT.get('PROJECT-NAME') or os.path.basename(PROJECT['PROJECT-ROOT'])
    sha = PROJECT['PROJECT-SHA']
    return f"{name} (git {sha[:12]})" if sha else name

def wrap(text):
    return textwrap.fill(text, 80, break_long_words=False, break_on_hyphens=False)

def agent_prompt(data, aborted):
    failed = collect_failures(data)
    if not failed:
        return None

    paths = [proj_relpath(n['suite_path'] if 'suite' in n else n['case']) for n in data['suite']]
    ran = ", ".join(f'"{p}"' for p in dict.fromkeys(paths))

    intro = (f"{len(failed)} test{'s' if len(failed) > 1 else ''} failed running {ran} "
             f"in project {project_label()} on host \"{os.uname().nodename}\", "
             f"log id {os.path.basename(LOGDIR)}.")
    if aborted:
        intro += " The run was aborted at the first fatal failure."
    lines = [wrap(intro), ""]

    for test in failed:
        if test.get('options'):
            called = f"called with arguments '{shlex.join(test['options'])}'"
        else:
            called = "called without arguments"
        lines.append(wrap(f"\"{test['name']}\" at \"{proj_relpath(test['case'])}\" "
                          f"{called} failed with:"))
        lines.append("```")
        for failure in test['failures']:
            lines.extend(failure)
            if len(failure) >= 40:
                lines.append("(more in the log)")
        lines += ["```", ""]

    lines.append("Please help me investigate this.")
    if PROJECT.get('AGENT-HINTS'):
        lines += ["", PROJECT['AGENT-HINTS'].strip()]
    return "\n".join(lines)

def probe_suite(data):
    for test in data['suite']:
        if 'suite' in test:
            probe_suite(test)
        elif 'case' in test:
                test['result'] = "noexec"
        else:
            print("error, garbage in suite")
            sys.exit(1)

    data['result'] = "noexec"

def count_suite_tree(data):
    suites = 0
    tests = 0
    for node in data['suite']:
        if 'suite' in node:
            suites += 1
            s, t = count_suite_tree(node)
            suites += s
            tests += t
        elif 'case' in node:
            tests += 1
    return suites, tests

def run_suite(args, data, skip_suite, fatal):
    skip = False
    err = False
    abort = False

    if data['name'] != "command-line":
        print(pcolor.blue + f"\nRunning suite {data['suite_num']}: {data['name']}" + pcolor.reset)

    for test in data['suite']:
        if 'suite' in test:
            subskip, suberr, subabort = run_suite(args, test, skip_suite,
                                                  fatal or test.get('abort'))
            if subskip:
                skip = True
            if suberr:
                err = True
            if subabort:
                abort = True
                break

        elif 'case' in test:
            if not os.path.isfile(test['case']):
                print(f"error, test case not found {test['case']}")
                sys.exit(1)
            if not os.access(test['case'], os.X_OK):
                print(f"error, test case not executable {test['case']}")
                sys.exit(1)

            if skip_suite:
                test['result'] = "skip"

            skip_suite, subskip, suberr = run_test(args, test)
            if suberr:
                if 'mask' in test and test['mask'] == "fail":
                    print(f"{pcolor.red}Test failure is masked in suite{pcolor.reset}")
                    test['result'] = "masked-fail"
                    err = False
                else:
                    test['result'] = "fail"
                    err = True

                if 'onfail' in test:
                    run_onfail(args, test)

                if err and (fatal or test.get('abort')):
                    print("Aborting execution")
                    abort = True
                    break
            elif subskip:
                if 'mask' in test and test['mask'] == "skip":
                    print(f"{pcolor.orange}Test skip is masked in suite{pcolor.reset}")
                    test['result'] = "masked-skip"
                else:
                    skip = True
                    test['result'] = "skip"
            else:
                test['result'] = "pass"

    if err:
        data['result'] = "fail"
    elif skip:
        data['result'] = "skip"
    else:
        data['result'] = "pass"

    return skip, err, abort

def get_first_existing_file(list, name):
    for f in list:
        if os.path.exists(os.path.expanduser(f)):
            return os.path.expanduser(f)
        vcprint(pcolor.faint, f"({name} not found: {f})")

    print("error, can't find any {name} to use")
    sys.exit(1)

def parse_proj_config(root_path, args):
    files = [
        os.path.join(root_path, '..', '9pm-proj.yaml'),
        os.path.join(root_path, 'etc', '9pm-proj.yaml')
    ]

    if "NINEPM_PROJ_CONFIG" in os.environ:
        files.insert(0, os.environ["NINEPM_PROJ_CONFIG"])

    if args.proj:
        files.insert(0, args.proj)

    path = get_first_existing_file(files, "Project Config")
    vcprint(pcolor.faint, f"Using project config: {path}")
    os.environ["NINEPM_PROJ_CONFIG"] = path

    try:
        with open(path, 'r') as f:
            data = yaml.safe_load(f) or {}
    except yaml.YAMLError:
        print(f"error, parsing YAML {path} config.")
        sys.exit(1)

    if not '9pm' in data:
        return {}

    if 'PROJECT-ROOT' in data['9pm']:
        data['9pm']['PROJECT-ROOT'] = rootify_path(data['9pm']['PROJECT-ROOT'])

    if 'PROJECT-TOPDOC' in data['9pm']:
        data['9pm']['PROJECT-TOPDOC'] = rootify_path(data['9pm']['PROJECT-TOPDOC'])

    return data['9pm']

def parse_rc(root_path, args):
    required_keys = ["LOG_PATH"]

    files = [
        os.path.join("~/.9pm.rc"),
        os.path.join(root_path, 'etc', '9pm.rc')
    ]

    path = get_first_existing_file(files, "Running Config")
    vcprint(pcolor.faint, f"Using RC: {path}")

    try:
        with open(path, 'r') as f:
            data = yaml.safe_load(f) or {}
    except yaml.YAMLError:
        print(f"error, parsing YAML {path} running config.")
        sys.exit(1)

    missing_keys = [key for key in required_keys if key not in data]
    if missing_keys:
        print(f"error, 9pm.rc is missing required keys: {', '.join(missing_keys)}")
        sys.exit(1)

    return data

def parse_cmdline():
    parser = argparse.ArgumentParser()
    parser.add_argument('-a', '--abort', action='store_true',
            help='(9PM) Abort execution if test fails')
    parser.add_argument('--no-exec', action='store_true',
            help='(9PM) Do not execute any tests')
    parser.add_argument('-p', '--proj', metavar='FILE', action='store',
            help='(9PM) Path to project configuration')
    parser.add_argument('-v', '--verbose', action='store_true',
            help='(9PM) Enable verbose output')
    parser.add_argument('-c', '--config', metavar='FILE', action='store',
            help='(TEST) Config file passed to test case')
    parser.add_argument('-d', '--debug', action='store_true',
            help='(TEST) Enable test case debug')
    parser.add_argument('-o', '--option', action='append', default=[],
            help='(TEST) Option(s) passed to all test cases (can be repeated). Use -o="-flag" for options starting with hyphens, like: -o="-ls" /bin/ls')
    parser.add_argument('-r', '--repeat', type=int, default=1,
            help='(TEST) Number of times to repeat the test (default: 1)')
    parser.add_argument('suites', nargs='+', metavar='TEST|SUITE',
            help='Test or suite to run')
    if len(sys.argv) == 1:
        # Normally, argparse does not display the help message if the user
        # didn't explicitly invoke '-h' but we also want it shown if the user
        # didn't specify any arguments at all.
        parser.print_help()
        sys.exit(1)
    return parser.parse_args()

def setup_log_dir(log_path):
    log_path = os.path.expanduser(log_path)

    now = datetime.now()
    dir_name = now.strftime('%Y-%m-%d_%H-%M-%S-%f')
    log_dir = os.path.join(log_path, dir_name)
    os.makedirs(log_dir)
    os.makedirs(os.path.join(log_dir, "output"))

    last_link =os.path.join(log_path, "last")
    if os.path.islink(last_link):
        os.unlink(last_link)
    os.symlink(dir_name, last_link)

    return log_dir

def setup_env(args):
    os.environ["NINEPM_TAP"] = "1"

    os.environ["NINEPM_ROOT_PATH"] = ROOT_PATH
    os.environ["NINEPM_DATABASE"] = DATABASE
    os.environ["NINEPM_SCRATCHDIR"] = SCRATCHDIR
    os.environ["NINEPM_LOG_PATH"] = LOGDIR

    if args.debug:
        os.environ["NINEPM_DEBUG"] = "1"
    if args.config:
        os.environ["NINEPM_CONFIG"] = args.config

def run_git_cmd(path, command):
    git_path = os.path.join(path, '.git')

    if os.path.isfile(git_path):
        with open(git_path, 'r') as f:
            line = f.read().strip()
            if line.startswith('gitdir: '):
                git_path = os.path.join(path, line[8:])
                if not os.path.exists(git_path):
                    return ""
            else:
                vcprint(pcolor.orange, f"warning, invalid .git file format ({path})")
                return ""

    if not os.path.isdir(git_path):
        vcprint(pcolor.orange, f"warning, no .git dir or file in path ({path})")
        return ""

    try:
        vcprint(pcolor.faint, f"Running: git --git-dir {git_path} {command}")
        result = subprocess.check_output(
            ['git', '--git-dir', git_path] + command,
            stderr=subprocess.STDOUT
        ).decode('utf-8').strip()
    except Exception as e:
        cprint(pcolor.orange, f"warning, git command failed ({e})")
        return ""

    return result

def create_base_suite(args):
    suite = {'name': 'command-line', 'suite': []}
    for _ in range(args.repeat):
        for filename in args.suites:
            fpath = os.path.join(os.getcwd(), filename)
            if filename.endswith('.yaml'):
                suite['suite'].append(parse_suite(fpath, "command-line", args.option, {}))
            else:
                test = {}
                test['case'] = fpath
                uniq_id, uname = gen_unix_name(filename)
                test['uniq_id'] = uniq_id
                test['unix_name'] = f"{uniq_id:04d}-{uname}"
                test['name'] = gen_name(filename)
                test['outfile'] = gen_outfile(test['unix_name'])

                if args.option:
                    test["options"] = args.option

                suite['suite'].append(test)
    return suite

def main():
    global DATABASE
    global SCRATCHDIR
    global LOGDIR
    global VERBOSE
    global NOEXEC
    global PROJECT

    # Line-buffer our own stdout so output streams when 9pm itself is piped to
    # another program (CI, tee, wrappers) instead of buffering until exit.
    sys.stdout.reconfigure(line_buffering=True)

    sha = ""
    if (sha := run_git_cmd(ROOT_PATH, ['rev-parse', 'HEAD'])):
        sha = f"({sha[:10]})"
    cprint(pcolor.yellow, f"9PM - Simplicity is the ultimate sophistication {sha}")

    args = parse_cmdline()
    VERBOSE = args.verbose
    NOEXEC = args.no_exec

    vcprint(pcolor.faint, f"Verbose output turned on")

    rc = parse_rc(ROOT_PATH, args)

    LOGDIR = setup_log_dir(rc['LOG_PATH'])

    vcprint(pcolor.faint, f"Logging to: {LOGDIR}")

    PROJECT = parse_proj_config(ROOT_PATH, args)
    PROJECT.setdefault('PROJECT-ROOT', os.path.dirname(ROOT_PATH))
    PROJECT['PROJECT-SHA'] = run_git_cmd(PROJECT['PROJECT-ROOT'], ['rev-parse', 'HEAD'])
    PROJECT['PROJECT-VERSION'] = run_git_cmd(PROJECT['PROJECT-ROOT'], ['describe', '--tags', '--always'])

    scratch = tempfile.mkdtemp(suffix='', prefix='9pm_', dir='/tmp')
    vcprint(pcolor.faint, f"Created scratch dir: {scratch}")
    SCRATCHDIR = scratch
    atexit.register(shutil.rmtree, SCRATCHDIR)

    db = tempfile.NamedTemporaryFile(suffix='_db', prefix='9pm_', dir=scratch)
    vcprint(pcolor.faint, f"Created databasefile: {db.name}")
    DATABASE = db.name

    cprint(pcolor.yellow, f"\nTesting {project_label()}")

    suite = create_base_suite(args)
    probe_suite(suite)

    n_suites, n_tests = count_suite_tree(suite)
    print(f"\nStarting: {n_suites} suites, {n_tests} tests")

    setup_env(args)

    skip, err, aborted = run_suite(args, suite, False, args.abort)
    if err:
        cprint(pcolor.red, "\nx Execution")
    elif skip:
        cprint(pcolor.yellow, "\ns Execution")
    else:
        cprint(pcolor.green, "\no Execution")

    print_result_tree(suite, "")

    prompt = agent_prompt(suite, aborted)
    if prompt:
        cprint(pcolor.yellow, "\nAGENT INITIAL INVESTIGATION PROMPT")
        print(prompt)
        with open(os.path.join(LOGDIR, 'agent-prompt.txt'), 'w') as f:
            f.write(prompt + "\n")

    # Export comprehensive JSON result
    json_path = write_json_result(suite, prompt)
    vcprint(pcolor.faint, f"JSON results written to: {json_path}")

    db.close()
    sys.exit(err)

if __name__ == '__main__':
    main()