Engineering Quality and Standards Questions

# Python runnable code
from collections import defaultdict, deque, namedtuple
Test = namedtuple("Test", ["id","start","end","resource"])

def cluster_by_contention(tests):
    # group tests by resource
    by_res = defaultdict(list)
    for t in tests:
        by_res[t.resource].append(t)

    clusters = []         # list of clusters (sets of test ids) across resources
    recommendations = []  # per-cluster advice

    for res, rtests in by_res.items():
        # sort by start time
        rtests.sort(key=lambda x: x.start)
        # build overlap graph using sweep line -> adjacency list
        adj = defaultdict(set)
        active = deque()  # store tests currently overlapping (by end time)
        for t in rtests:
            # pop non-overlapping
            while active and active[0].end <= t.start:
                active.popleft()
            # any remaining in active overlap with t
            for a in active:
                adj[t.id].add(a.id); adj[a.id].add(t.id)
            active.append(t)
        # extract connected components
        seen = set()
        for t in rtests:
            if t.id in seen: continue
            q=[t.id]; comp=set()
            while q:
                u=q.pop()
                if u in comp: continue
                comp.add(u)
                seen.add(u)
                for v in adj[u]:
                    if v not in comp: q.append(v)
            if len(comp)>1:
                clusters.append((res,comp))
                # compute simple metrics
                overlaps = sum(len(adj[x]) for x in comp)/len(comp)
                span = (max(next(tt for tt in rtests if tt.id==x).end for x in comp)
                        - min(next(tt for tt in rtests if tt.id==x).start for x in comp))
                avg_dur = sum(next(tt for tt in rtests if tt.id==x).end - next(tt for tt in rtests if tt.id==x).start for x in comp)/len(comp)
                score = overlaps * 0.6 + (span/ (avg_dur+1)) * 0.4
                if overlaps>1.0 or span/avg_dur>2:
                    rec = f"Resource {res} cluster {comp}: high contention. Suggest: shard tests across resources, add resource pool, or set affinity to distribute; where impossible, serialize small groups or increase resource parallelism."
                else:
                    rec = f"Resource {res} cluster {comp}: moderate contention. Suggest: small delays (jitter), limit parallelism per resource."
                recommendations.append((comp,score,rec))
    return clusters, sorted(recommendations, key=lambda x:-x[1])