Function test_step_memory

tests/integration/manual_integration_check.py:450–493

(dir, days)

def test_step_memory(dir, days):
    days_since = int(days)
    sacct_start_date = (datetime.now() - timedelta(days=days_since)).strftime(
        "%Y-%m-%d"
    )
    sacct_command = f"sacct --starttime {sacct_start_date} -u {os.environ['USER']} --noheader -o jobid,jobname%80,AllocTRES%80,ElapsedRaw,stat%30,NodeList,Start,MaxRSS"
    print(f"Running + {sacct_command}")
    sacct_output = subprocess.check_output(sacct_command.split()).decode()
    relevant_started_jobs = []
    mem_dict = dict(re.findall(r"([0-9]+)\.0 +.* +([0-9]*.?[0-9]*[KMG])", sacct_output))

    completed_seeds = os.path.join(dir, "finished_seeds.txt")
    seeds = open(completed_seeds).read()

    for sacct_line in sacct_output.splitlines():
        parsed_job = parse_sacct_line(sacct_line)
        if parsed_job is None:
            continue
        for name in re.findall(f"{Path(dir).stem}_([^ _]+)_.*", parsed_job.name):
            if name in seeds:
                if parsed_job.job_id in mem_dict:
                    max_memory = mem_dict[parsed_job.job_id]
                    parsed_job.max_memory_gb = (
                        float(max_memory[:-1]) / MEM_FACTOR[max_memory[-1]]
                    )
                relevant_started_jobs.append(parsed_job)

    step_mem = {
        "fineterrain": [],
        "coarse": [],
        "populate": [],
        "rendershort": [],
        "blendergt": [],
    }

    for job in relevant_started_jobs:
        for step in step_mem:
            if step in job.name:
                step_mem[step].append(job.max_memory_gb)

    step_mem_df = pd.DataFrame.from_dict(step_mem, orient="index")
    step_mem_stats = make_stats(step_mem_df)
    print("\nMemory Usage by Step")
    print(tabulate(step_mem_stats, headers="keys", tablefmt="fancy_grid"))


def test_brightness(dir):