zzstoatzz.io
This repository has no description
1#!/usr/bin/env -S uv run --script --quiet
2# /// script
3# requires-python = ">=3.12"
4# dependencies = ["prefect>=3.0"]
5# ///
6"""smoke test: drive a real prefect client against weir.
7
8usage:
9 PREFECT_API_URL=http://localhost:4200/api ./scripts/test-flow
10"""
11import os
12import tempfile
13from pathlib import Path
14
15from prefect import flow, task
16
17RESULT_DIR = Path(tempfile.mkdtemp(prefix="weir-test-"))
18
19
20@task
21def add_task(a: int, b: int) -> int:
22 return a + b
23
24
25@task
26def multiply_task(a: int, b: int) -> int:
27 return a * b
28
29
30@flow
31def basic_flow(a: int, b: int) -> int:
32 sum_result = add_task(a, b)
33 product = multiply_task(sum_result, 2)
34 return product
35
36
37def main():
38 os.environ.setdefault("PREFECT_LOCAL_STORAGE_PATH", str(RESULT_DIR))
39 print(f"api url: {os.environ.get('PREFECT_API_URL', '(not set)')}")
40
41 result = basic_flow(3, 4)
42 assert result == 14, f"expected 14, got {result}"
43 print(f"basic_flow(3, 4) = {result}")
44 print("=== flow ran against weir ===")
45
46
47if __name__ == "__main__":
48 main()