Problem Solving and Communication Approach Questions

from collections import defaultdict
import math, heapq

def correlate_failures(records, top_n=50, min_support=2, measure='phi'):
    # counts
    total_events = 0
    test_failures = defaultdict(int)   # test -> fail count
    test_totals = defaultdict(int)     # test -> total count
    file_changes = defaultdict(int)    # file -> times changed
    pair_cf_fail = defaultdict(int)    # (file,test) -> changed & failed
    pair_cf_pass = defaultdict(int)    # (file,test) -> changed & passed

    for ev in records:
        total_events += 1
        test = ev['test_id']
        status = ev['status']  # 'FAIL' or 'PASS'
        changed = ev.get('changed_files', []) or []
        test_totals[test] += 1
        if status != 'PASS':
            test_failures[test] += 1

        for f in changed:
            file_changes[f] += 1
            if status != 'PASS':
                pair_cf_fail[(f, test)] += 1
            else:
                pair_cf_pass[(f, test)] += 1

    # score pairs, keep top-N
    heap = []
    for (f, t), a in pair_cf_fail.items():
        b = pair_cf_pass.get((f, t), 0)                # changed & passed
        ft_fail = test_failures[t]
        ft_total = test_totals[t]
        # derived counts:
        A = a                                          # changed & failed
        B = b                                          # changed & passed
        C = ft_fail = ft_fail = ft_fail = ft_fail if False else (ft_fail := test_failures[t])  # placeholder for clarity

    # compute properly outside broken construct
    heap = []
    for (f, t), A in pair_cf_fail.items():
        B = pair_cf_pass.get((f, t), 0)
        FT_fail = test_failures[t]
        FT_total = test_totals[t]
        if A + B < min_support:
            continue
        # contingency:
        # A = changed & failed
        # B = changed & passed
        # C = not_changed & failed = FT_fail - A
        # D = not_changed & passed = (FT_total - FT_fail) - B
        C = FT_fail - A
        D = (FT_total - FT_fail) - B
        # phi coefficient:
        denom = math.sqrt((A+B)*(C+D)*(A+C)*(B+D))
        phi = (A*D - B*C) / denom if denom > 0 else 0.0
        # lift = P(f changed & fail) / (P(f changed)*P(fail))
        p_f_changed = (A + B) / total_events
        p_fail = FT_fail / total_events
        p_both = A / total_events
        lift = (p_both / (p_f_changed * p_fail)) if p_f_changed > 0 and p_fail > 0 else 0.0

        score = phi if measure == 'phi' else lift
        if len(heap) < top_n:
            heapq.heappush(heap, (score, f, t, A, B))
        else:
            heapq.heappushpop(heap, (score, f, t, A, B))

    # return sorted descending
    return sorted(heap, key=lambda x: -x[0])

from collections import defaultdict

def find_flaky_tests(events, min_runs=5, treat_unstable_as_fail=True):
    # events: iterable of (test_id, build_id, status) assumed chronological per test
    buckets = defaultdict(list)
    for test_id, build_id, status in events:
        s = status.upper()
        if s == "UNSTABLE":
            s = "FAIL" if treat_unstable_as_fail else None
        if s is None:
            continue
        buckets[test_id].append(s)

    result = {}
    for test_id, seq in buckets.items():
        n = len(seq)
        if n < min_runs:
            continue
        flips = sum(1 for i in range(1, n) if seq[i] != seq[i-1])
        score = flips / (n - 1) if n > 1 else 0.0
        if flips > 0:
            result[test_id] = {"runs": n, "flips": flips, "score": score}
    return result