File size: 2,170 Bytes
4d9b003 | 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 | # SPDX-License-Identifier: Apache-2.0
"""pytest fixtures of diffusion-planner-p150 (no dependency on tt-metal's own conftest).
- ``device`` (session): one chip opened like the published numbers (``tt_diffusion_planner.device.open_device``:
ETH dispatch, 12x10 grid, the port's validated sizes and CQs). ``--device-id N`` or ``TT_DEVICE_ID`` selects
the chip (default 0); ``DIFFUSION_PLANNER_DISPATCH=worker`` is the A/B opt-in, and the other ``DIFFUSION_PLANNER_*``
variables apply as for the server. A failing ETH open is an error here, never a silent WORKER fallback: gates
are only valid on the published setup.
- Tests marked ``device`` are skipped when ttnn is missing or ``TT_VISIBLE_DEVICES=none`` (host-only runs).
On the shared workspace box run them through the lock and name the test files:
``bin/devrun -t 1800 -- python -m pytest -q -s code/tt_diffusion_planner/tests/test_pcc_device.py``.
"""
from __future__ import annotations
import gc
import importlib.util
import os
import pytest
def pytest_addoption(parser):
parser.addoption("--device-id", action="store", default=None, help="chip id (default $TT_DEVICE_ID or 0)")
def pytest_configure(config):
config.addinivalue_line("markers", "device: needs a Tenstorrent chip (skipped on host-only runs)")
def _no_device_reason():
if os.environ.get("TT_VISIBLE_DEVICES", "").lower() == "none":
return "TT_VISIBLE_DEVICES=none (host-only run)"
if importlib.util.find_spec("ttnn") is None:
return "ttnn is not installed"
return None
def pytest_collection_modifyitems(config, items):
reason = _no_device_reason()
if reason:
skip = pytest.mark.skip(reason=reason)
for item in items:
if "device" in item.keywords:
item.add_marker(skip)
@pytest.fixture(autouse=True)
def _gc_between_tests():
gc.collect()
@pytest.fixture(scope="session")
def device(request):
from tt_diffusion_planner.device import close_device, open_device
cli = request.config.getoption("--device-id")
dev = open_device(int(cli) if cli is not None else None, allow_fallback=False)
yield dev
close_device(dev)
|