()
| 18 | |
| 19 | |
| 20 | def test_deploy(): |
| 21 | tmp_dir = Path(tempfile.mkdtemp()) |
| 22 | runner_dir = tmp_dir / "runner" |
| 23 | runner_dir.mkdir() |
| 24 | |
| 25 | try: |
| 26 | subprocess.check_call( |
| 27 | [ |
| 28 | uv.find_uv_bin(), |
| 29 | "run", |
| 30 | "--isolated", |
| 31 | "prefect", |
| 32 | "work-pool", |
| 33 | "create", |
| 34 | "test-deploy-pool", |
| 35 | "-t", |
| 36 | "process", |
| 37 | ], |
| 38 | stdout=sys.stdout, |
| 39 | stderr=sys.stderr, |
| 40 | ) |
| 41 | |
| 42 | # Create GitRepository and set base path to temp directory |
| 43 | # to avoid race conditions with parallel tests |
| 44 | git_repo = GitRepository( |
| 45 | url="https://github.com/PrefectHQ/examples.git", |
| 46 | ) |
| 47 | git_repo.set_base_path(tmp_dir) |
| 48 | |
| 49 | flow_instance = prefect.flow.from_source( |
| 50 | source=git_repo, |
| 51 | entrypoint="flows/hello_world.py:hello", |
| 52 | ) |
| 53 | |
| 54 | flow_instance.deploy( |
| 55 | name="demo-deployment", |
| 56 | work_pool_name="test-deploy-pool", |
| 57 | parameters={"name": "world"}, |
| 58 | ) |
| 59 | |
| 60 | flow_run = run_deployment("hello/demo-deployment", timeout=0) |
| 61 | |
| 62 | subprocess.check_call( |
| 63 | [ |
| 64 | uv.find_uv_bin(), |
| 65 | "run", |
| 66 | "--isolated", |
| 67 | "prefect", |
| 68 | "flow-run", |
| 69 | "execute", |
| 70 | str(flow_run.id), |
| 71 | ], |
| 72 | stdout=sys.stdout, |
| 73 | stderr=sys.stderr, |
| 74 | cwd=runner_dir, |
| 75 | ) |
| 76 | |
| 77 | flow_run = anyio.run(read_flow_run, flow_run.id) |
nothing calls this directly
no test coverage detected
searching dependent graphs…