Continuous Integration and Delivery Pipeline Testing Questions

apiVersion: release.example.com/v1
kind: ReleaseMetadata
metadata:
  name: myservice-20251122-abcdef
spec:
  image: registry.example.com/myservice@sha256:abcdef...
  tests:
    unit: passed
    integration: passed
    e2e: passed
  signedBy: ci-system@example.com
  createdAt: "2025-11-22T12:34:56Z"

#!/usr/bin/env python3
"""
List top-level packages changed between HEAD and origin/main.
Outputs one package per line; use '-' for repo root changes.
"""
import subprocess
import sys
from pathlib import PurePosixPath

def git_changed_files(base='origin/main'):
    # Uses git diff --name-only to list changed files between HEAD and base
    out = subprocess.run(['git', 'fetch', '--no-tags', '--quiet', 'origin', 'main'],
                         stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    proc = subprocess.run(['git', 'diff', '--name-only', f'{base}..HEAD'],
                          stdout=subprocess.PIPE, text=True, check=True)
    return [line.strip() for line in proc.stdout.splitlines() if line.strip()]

def top_level_packages(paths):
    pkgs = set()
    for p in paths:
        pp = PurePosixPath(p)
        parts = pp.parts
        if not parts:
            continue
        # if file at repo root -> denote as repo-root
        if len(parts) == 1:
            pkgs.add(parts[0])
        else:
            pkgs.add(parts[0])
    return sorted(pkgs)

def main():
    try:
        files = git_changed_files()
    except subprocess.CalledProcessError as e:
        print("Error running git:", e, file=sys.stderr)
        sys.exit(2)
    if not files:
        return
    pkgs = top_level_packages(files)
    for p in pkgs:
        print(p)

if __name__ == '__main__':
    main()

import heapq
from typing import List, Tuple, Dict

def shard_tests(tests: List[Tuple[str, float]], K: int) -> Dict[int, List[Tuple[str, float]]]:
    """
    Assign tests to K shards balancing cumulative durations.
    Returns dict: shard_index -> list of (test_name, duration)
    Deterministic: sorts by (-duration, name) and uses shard index tie-breakers.
    """
    if K <= 0:
        raise ValueError("K must be >= 1")
    # Initialize empty shards
    shards = {i: [] for i in range(K)}
    # Sort by descending duration, deterministic tie-breaker by name
    tests_sorted = sorted(tests, key=lambda t: (-t[1], t[0]))
    # Min-heap of (current_total_duration, shard_index)
    heap = [(0.0, i) for i in range(K)]
    heapq.heapify(heap)
    for name, dur in tests_sorted:
        total, idx = heapq.heappop(heap)             # lightest shard
        shards[idx].append((name, dur))              # assign test
        heapq.heappush(heap, (total + float(dur), idx))
    return shards

# Example usage:
# tests = [("t1", 10), ("t2", 3), ("t3", 7), ("t4", 4)]
# print(shard_tests(tests, 2))