if not is_torch_available():

raise ValueError("PyTorch needs to be installed to use this function.")

if not torch.allclose(a, b, *args, **kwargs):

assert False, f"Max diff is absolute {(a - b).abs().max()}. Diff tensor is {(a - b).abs()}."

test_name = os.environ.get("PYTEST_CURRENT_TEST")

if not torch.is_tensor(tensor):

tensor = torch.from_numpy(tensor)

tensor_str = str(tensor.detach().cpu().flatten().to(torch.float32)).replace("\n", "")

# format is usually:

# expected_slice = np.array([-0.5713, -0.3018, -0.9814, 0.04663, -0.879, 0.76, -1.734, 0.1044, 1.161])

output_str = tensor_str.replace("tensor", f"{expected_tensor_name} = np.array")

test_file, test_class, test_fn = test_name.split("::")

test_fn = test_fn.split()[0]

with open(filename, "a") as f:

"""

# this function caller's __file__

caller__file__ = inspect.stack()[1][1]

tests_dir = os.path.abspath(os.path.dirname(caller__file__))

while not tests_dir.endswith("tests"):

tests_dir = os.path.dirname(tests_dir)

if append_path:

return os.path.join(tests_dir, append_path)

else:

try:

value = os.environ[key]

except KeyError:

# KEY isn't set, default to `default`.

_value = default

else:

# KEY is set, convert it to True or False.

try:

_value = strtobool(value)

except ValueError:

# More values are supported, but let's keep the message simple.

raise ValueError(f"If set, {key} must be yes or no.")

"""Creates a random float32 tensor"""

if rng is None:

rng = global_rng

total_dims = 1

for dim in shape:

total_dims *= dim

values = []

for _ in range(total_dims):

values.append(rng.random() * scale)

"""

"""

"""

"""

return unittest.skipUnless(is_torch_available() and is_torch_version(">=", "2.0.0"), "test requires PyTorch 2")(

test_case

"""Decorator marking a test that requires CUDA and PyTorch."""

return unittest.skipUnless(is_torch_available() and torch_device == "cuda", "test requires PyTorch+CUDA")(

test_case

"""

"""

"""

"""

"""

if isinstance(arry, str):

# local_path = "/home/patrick_huggingface_co/"

if local_path is not None:

# local_path can be passed to correct images of tests

return os.path.join(local_path, "/".join([arry.split("/")[-5], arry.split("/")[-2], arry.split("/")[-1]]))

elif arry.startswith("http://") or arry.startswith("https://"):

response = requests.get(arry)

response.raise_for_status()

arry = np.load(BytesIO(response.content))

elif os.path.isfile(arry):

arry = np.load(arry)

else:

raise ValueError(

f"Incorrect path or url, URLs must start with `http://` or `https://`, and {arry} is not a valid path"

)

elif isinstance(arry, np.ndarray):

pass

else:

raise ValueError(

"Incorrect format used for numpy ndarray. Should be an url linking to an image, a local path, or a"

" ndarray."

)

response = requests.get(url)

response.raise_for_status()

arry = torch.load(BytesIO(response.content))

"""

if isinstance(image, str):

if image.startswith("http://") or image.startswith("https://"):

image = PIL.Image.open(requests.get(image, stream=True).raw)

elif os.path.isfile(image):

image = PIL.Image.open(image)

else:

raise ValueError(

f"Incorrect path or url, URLs must start with `http://` or `https://`, and {image} is not a valid path"

)

elif isinstance(image, PIL.Image.Image):

image = image

else:

raise ValueError(

"Incorrect format used for image. Should be an url linking to an image, a local path, or a PIL image."

)

image = PIL.ImageOps.exif_transpose(image)

image = image.convert("RGB")

w, h = image.size

w, h = (x - x % 8 for x in (w, h)) # resize to integer multiple of 8

image = image.resize((w, h), resample=PIL.Image.LANCZOS)

image = np.array(image).astype(np.float32) / 255.0

image = np.vstack([image[None].transpose(0, 3, 1, 2)] * batch_size)

image = torch.from_numpy(image)

if is_opencv_available():

import cv2

else:

raise ImportError(BACKENDS_MAPPING["opencv"][1].format("export_to_video"))

if output_video_path is None:

output_video_path = tempfile.NamedTemporaryFile(suffix=".mp4").name

fourcc = cv2.VideoWriter_fourcc(*"mp4v")

h, w, c = video_frames[0].shape

video_writer = cv2.VideoWriter(output_video_path, fourcc, fps=8, frameSize=(w, h))

for i in range(len(video_frames)):

img = cv2.cvtColor(video_frames[i], cv2.COLOR_RGB2BGR)

video_writer.write(img)

if not path.startswith("http://") or path.startswith("https://"):

path = os.path.join(

"https://huggingface.co/datasets/fusing/diffusers-testing/resolve/main", urllib.parse.quote(path)

)

"""

option = "--make-reports"

if option not in pytest_opt_registered:

parser.addoption(

option,

action="store",

default=False,

help="generate report files. The value of this option is used as a prefix to report names",

)

"""

from _pytest.config import create_terminal_writer

if not len(id):

id = "tests"

config = tr.config

orig_writer = config.get_terminal_writer()

orig_tbstyle = config.option.tbstyle

orig_reportchars = tr.reportchars

dir = "reports"

Path(dir).mkdir(parents=True, exist_ok=True)

report_files = {

k: f"{dir}/{id}_{k}.txt"

for k in [

"durations",

"errors",

"failures_long",

"failures_short",

"failures_line",

"passes",

"stats",

"summary_short",

"warnings",

]

}

# custom durations report

# note: there is no need to call pytest --durations=XX to get this separate report

# adapted from https://github.com/pytest-dev/pytest/blob/897f151e/src/_pytest/runner.py#L66

dlist = []

for replist in tr.stats.values():

for rep in replist:

if hasattr(rep, "duration"):

dlist.append(rep)

if dlist:

dlist.sort(key=lambda x: x.duration, reverse=True)

with open(report_files["durations"], "w") as f:

durations_min = 0.05 # sec

f.write("slowest durations\n")

for i, rep in enumerate(dlist):

if rep.duration < durations_min:

f.write(f"{len(dlist)-i} durations < {durations_min} secs were omitted")

break

f.write(f"{rep.duration:02.2f}s {rep.when:<8} {rep.nodeid}\n")

def summary_failures_short(tr):

# expecting that the reports were --tb=long (default) so we chop them off here to the last frame

reports = tr.getreports("failed")

if not reports:

return

tr.write_sep("=", "FAILURES SHORT STACK")

for rep in reports:

msg = tr._getfailureheadline(rep)

tr.write_sep("_", msg, red=True, bold=True)

# chop off the optional leading extra frames, leaving only the last one

longrepr = re.sub(r".*_ _ _ (_ ){10,}_ _ ", "", rep.longreprtext, 0, re.M | re.S)

tr._tw.line(longrepr)

# note: not printing out any rep.sections to keep the report short

# use ready-made report funcs, we are just hijacking the filehandle to log to a dedicated file each

# adapted from https://github.com/pytest-dev/pytest/blob/897f151e/src/_pytest/terminal.py#L814

# note: some pytest plugins may interfere by hijacking the default `terminalreporter` (e.g.

# pytest-instafail does that)

# report failures with line/short/long styles

config.option.tbstyle = "auto" # full tb

with open(report_files["failures_long"], "w") as f:

tr._tw = create_terminal_writer(config, f)

tr.summary_failures()

# config.option.tbstyle = "short" # short tb

with open(report_files["failures_short"], "w") as f:

tr._tw = create_terminal_writer(config, f)

summary_failures_short(tr)

config.option.tbstyle = "line" # one line per error

with open(report_files["failures_line"], "w") as f:

tr._tw = create_terminal_writer(config, f)

tr.summary_failures()

with open(report_files["errors"], "w") as f:

tr._tw = create_terminal_writer(config, f)

tr.summary_errors()

with open(report_files["warnings"], "w") as f:

tr._tw = create_terminal_writer(config, f)

tr.summary_warnings() # normal warnings

tr.summary_warnings() # final warnings

tr.reportchars = "wPpsxXEf" # emulate -rA (used in summary_passes() and short_test_summary())

with open(report_files["passes"], "w") as f:

tr._tw = create_terminal_writer(config, f)

tr.summary_passes()

with open(report_files["summary_short"], "w") as f:

tr._tw = create_terminal_writer(config, f)

tr.short_test_summary()

with open(report_files["stats"], "w") as f:

tr._tw = create_terminal_writer(config, f)

tr.summary_stats()

# restore:

tr._tw = orig_writer

tr.reportchars = orig_reportchars

"""

def __init__(self, logger):

self.logger = logger

self.io = StringIO()

self.sh = logging.StreamHandler(self.io)

self.out = ""

def __enter__(self):

self.logger.addHandler(self.sh)

return self

def __exit__(self, *exc):

self.logger.removeHandler(self.sh)

self.out = self.io.getvalue()

def __repr__(self):