liner_broker 1.3.0

Redis based message serverless broker.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

#!/usr/bin/python3
# -*- coding: utf-8 -*-

"""
Run SQLite-backed integration scripts under ``test/sqlite/``.

Usage:
  python3 test/sqlite/run_integration.py
  python3 test/sqlite/run_integration.py --list
  python3 test/sqlite/run_integration.py --only burst,offline
  python3 test/sqlite/run_integration.py --continue-on-fail

Cooperating processes share one temporary ``*.sqlite`` file per script (no Redis).
"""

from __future__ import annotations

import argparse
import os
import subprocess
import sys
import time
from pathlib import Path


ROOT = Path(__file__).resolve().parent
REPO = ROOT.parent.parent


def discover() -> list[Path]:
    scripts: list[Path] = []
    scripts.extend(sorted(ROOT.glob("integration_*.py")))
    for name in [
        "offline_delivery.py",
        "offline_delivery_simple.py",
        "offline_delivery_more.py",
    ]:
        p = ROOT / name
        if p.exists():
            scripts.append(p)
    seen: set[Path] = set()
    out: list[Path] = []
    for p in scripts:
        if p not in seen:
            seen.add(p)
            out.append(p)
    return out


def short_name(path: Path) -> str:
    if path.name.startswith("integration_"):
        return path.name.removeprefix("integration_").removesuffix(".py")
    return path.name.removesuffix(".py")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--list", action="store_true", help="List discovered integration scripts")
    ap.add_argument(
        "--only",
        default="",
        help="Comma-separated substrings to filter tests by name",
    )
    ap.add_argument(
        "--continue-on-fail",
        action="store_true",
        help="Run all tests even if some fail",
    )
    args = ap.parse_args()

    scripts = discover()
    if not scripts:
        print("No integration scripts found.", file=sys.stderr)
        return 2

    if args.list:
        for p in scripts:
            print(short_name(p))
        return 0

    filters: list[str] = [f.strip() for f in args.only.split(",") if f.strip()]
    if filters:
        scripts = [p for p in scripts if any(f in short_name(p) for f in filters)]

    if not scripts:
        print("No scripts matched --only filter.", file=sys.stderr)
        return 2

    failures: list[tuple[str, int]] = []
    start_all = time.time()
    for p in scripts:
        name = short_name(p)
        print(f"\n=== RUN {name} (sqlite) ===", flush=True)
        start = time.time()
        proc = subprocess.run(
            [sys.executable, str(p)],
            cwd=str(REPO),
            env=os.environ.copy(),
        )
        elapsed = time.time() - start
        if proc.returncode == 0:
            print(f"=== OK  {name} ({elapsed:.2f}s) ===", flush=True)
        else:
            print(
                f"=== FAIL {name} rc={proc.returncode} ({elapsed:.2f}s) ===",
                file=sys.stderr,
                flush=True,
            )
            failures.append((name, proc.returncode))
            if not args.continue_on_fail:
                break

    elapsed_all = time.time() - start_all
    if failures:
        print("\nFailed tests:", file=sys.stderr)
        for n, rc in failures:
            print(f"- {n}: rc={rc}", file=sys.stderr)
        print(f"\nTOTAL: FAIL ({elapsed_all:.2f}s)", file=sys.stderr)
        return 1

    print(f"\nTOTAL: OK ({elapsed_all:.2f}s)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())