File: client.py

package info (click to toggle)
python-websockets 15.0.1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 2,948 kB
  • sloc: python: 25,105; javascript: 350; ansic: 148; makefile: 43
file content (58 lines) | stat: -rw-r--r-- 1,272 bytes parent folder | download
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
import json
import logging

from websockets.exceptions import WebSocketException
from websockets.sync.client import connect


logging.basicConfig(level=logging.WARNING)

SERVER = "ws://localhost:9001"

AGENT = "websockets.sync"


def get_case_count():
    with connect(f"{SERVER}/getCaseCount") as ws:
        return json.loads(ws.recv())


def run_case(case):
    with connect(
        f"{SERVER}/runCase?case={case}&agent={AGENT}",
        max_size=2**25,
    ) as ws:
        try:
            for msg in ws:
                ws.send(msg)
        except WebSocketException:
            pass


def update_reports():
    with connect(
        f"{SERVER}/updateReports?agent={AGENT}",
        open_timeout=60,
    ):
        pass


def main():
    cases = get_case_count()
    for case in range(1, cases + 1):
        print(f"Running test case {case:03d} / {cases}... ", end="\t")
        try:
            run_case(case)
        except WebSocketException as exc:
            print(f"ERROR: {type(exc).__name__}: {exc}")
        except Exception as exc:
            print(f"FAIL: {type(exc).__name__}: {exc}")
        else:
            print("OK")
    print(f"Ran {cases} test cases")
    update_reports()
    print("Updated reports")


if __name__ == "__main__":
    main()