"controlnet",

"stable_diffusion",

"stable_diffusion_2",

"stable_diffusion_xl",

"stable_video_diffusion",

"deepfloyd_if",

"kandinsky",

"kandinsky2_2",

"text_to_video_synthesis",

"wuerstchen",

"""

current_head = repo.head.commit if repo.head.is_detached else repo.head.ref

try:

repo.git.checkout(commit_id)

yield

finally:

"""

# We need to deactivate autoformatting here to write escaped triple quotes (we cannot use real triple quotes or

# this would mess up the result if this function applied to this particular file).

# fmt: off

# Remove docstrings by splitting on triple " then triple ':

splits = content.split('\"\"\"')

content = "".join(splits[::2])

splits = content.split("\'\'\'")

# fmt: on

content = "".join(splits[::2])

# Remove empty lines and comments

lines_to_keep = []

for line in content.split("\n"):

# remove anything that is after a # sign.

line = re.sub("#.*$", "", line)

# remove white lines

if len(line) != 0 and not line.isspace():

lines_to_keep.append(line)

"""

# Keep doc examples only by splitting on triple "`"

splits = content.split("```")

# Add leading and trailing "```" so the navigation is easier when compared to the original input `content`

content = "```" + "```".join(splits[1::2]) + "```"

# Remove empty lines and comments

lines_to_keep = []

for line in content.split("\n"):

# remove anything that is after a # sign.

line = re.sub("#.*$", "", line)

# remove white lines

if len(line) != 0 and not line.isspace():

lines_to_keep.append(line)

"""

# test folders/files directly under `tests` folder

tests = os.listdir(PATH_TO_TESTS)

tests = [f"tests/{f}" for f in tests if "__pycache__" not in f]

tests = sorted([f for f in tests if (PATH_TO_REPO / f).is_dir() or f.startswith("tests/test_")])

"""

folder = Path(repo.working_dir)

with checkout_commit(repo, branching_point):

with open(folder / filename, "r", encoding="utf-8") as f:

old_content = f.read()

with open(folder / filename, "r", encoding="utf-8") as f:

new_content = f.read()

old_content_clean = clean_code(old_content)

new_content_clean = clean_code(new_content)

"""

folder = Path(repo.working_dir)

with checkout_commit(repo, branching_point):

with open(folder / filename, "r", encoding="utf-8") as f:

old_content = f.read()

with open(folder / filename, "r", encoding="utf-8") as f:

new_content = f.read()

old_content_clean = keep_doc_examples_only(old_content)

new_content_clean = keep_doc_examples_only(new_content)

"""

print("\n### DIFF ###\n")

code_diff = []

for commit in commits:

for diff_obj in commit.diff(base_commit):

# We always add new python files

if diff_obj.change_type == "A" and diff_obj.b_path.endswith(".py"):

code_diff.append(diff_obj.b_path)

# We check that deleted python files won't break corresponding tests.

elif diff_obj.change_type == "D" and diff_obj.a_path.endswith(".py"):

code_diff.append(diff_obj.a_path)

# Now for modified files

elif diff_obj.change_type in ["M", "R"] and diff_obj.b_path.endswith(".py"):

# In case of renames, we'll look at the tests using both the old and new name.

if diff_obj.a_path != diff_obj.b_path:

code_diff.extend([diff_obj.a_path, diff_obj.b_path])

else:

# Otherwise, we check modifications are in code and not docstrings.

if diff_is_docstring_only(repo, commit, diff_obj.b_path):

print(f"Ignoring diff in {diff_obj.b_path} as it only concerns docstrings or comments.")

else:

code_diff.append(diff_obj.a_path)

"""

repo = Repo(PATH_TO_REPO)

if not diff_with_last_commit:

# Need to fetch refs for main using remotes when running with github actions.

upstream_main = repo.remotes.origin.refs.main

print(f"main is at {upstream_main.commit}")

print(f"Current head is at {repo.head.commit}")

branching_commits = repo.merge_base(upstream_main, repo.head)

for commit in branching_commits:

print(f"Branching commit: {commit}")

return get_diff(repo, repo.head.commit, branching_commits)

else:

print(f"main is at {repo.head.commit}")

parent_commits = repo.head.commit.parents

for commit in parent_commits:

print(f"Parent commit: {commit}")

"""

print("\n### DIFF ###\n")

code_diff = []

for commit in commits:

for diff_obj in commit.diff(base_commit):

# We only consider Python files and doc files.

if not diff_obj.b_path.endswith(".py") and not diff_obj.b_path.endswith(".md"):

continue

# We always add new python/md files

if diff_obj.change_type in ["A"]:

code_diff.append(diff_obj.b_path)

# Now for modified files

elif diff_obj.change_type in ["M", "R"]:

# In case of renames, we'll look at the tests using both the old and new name.

if diff_obj.a_path != diff_obj.b_path:

code_diff.extend([diff_obj.a_path, diff_obj.b_path])

else:

# Otherwise, we check modifications contain some doc example(s).

if diff_contains_doc_examples(repo, commit, diff_obj.b_path):

code_diff.append(diff_obj.a_path)

else:

print(f"Ignoring diff in {diff_obj.b_path} as it doesn't contain any doc example.")

"""

py_files = [str(x.relative_to(PATH_TO_REPO)) for x in PATH_TO_REPO.glob("**/*.py")]

md_files = [str(x.relative_to(PATH_TO_REPO)) for x in PATH_TO_REPO.glob("**/*.md")]

test_files_to_run = py_files + md_files

# only include files in `src` or `docs/source/en/`

test_files_to_run = [x for x in test_files_to_run if x.startswith(("src/", "docs/source/en/"))]

# not include init files

test_files_to_run = [x for x in test_files_to_run if not x.endswith(("__init__.py",))]

# These are files not doctested yet.

with open("utils/not_doctested.txt") as fp:

not_doctested = {x.split(" ")[0] for x in fp.read().strip().split("\n")}

# So far we don't have 100% coverage for doctest. This line will be removed once we achieve 100%.

test_files_to_run = [x for x in test_files_to_run if x not in not_doctested]

"""

for diff_obj in branching_commit.diff(base_commit):

# Ignores all but the "utils/not_doctested.txt" file.

if diff_obj.a_path != "utils/not_doctested.txt":

continue

# Loads the two versions

folder = Path(repo.working_dir)

with checkout_commit(repo, branching_commit):

with open(folder / "utils/not_doctested.txt", "r", encoding="utf-8") as f:

old_content = f.read()

with open(folder / "utils/not_doctested.txt", "r", encoding="utf-8") as f:

new_content = f.read()

# Compute the removed lines and return them

removed_content = {x.split(" ")[0] for x in old_content.split("\n")} - {

x.split(" ")[0] for x in new_content.split("\n")

}

return sorted(removed_content)

"""

repo = Repo(PATH_TO_REPO)

test_files_to_run = [] # noqa

if not diff_with_last_commit:

upstream_main = repo.remotes.origin.refs.main

print(f"main is at {upstream_main.commit}")

print(f"Current head is at {repo.head.commit}")

branching_commits = repo.merge_base(upstream_main, repo.head)

for commit in branching_commits:

print(f"Branching commit: {commit}")

test_files_to_run = get_diff_for_doctesting(repo, repo.head.commit, branching_commits)

else:

print(f"main is at {repo.head.commit}")

parent_commits = repo.head.commit.parents

for commit in parent_commits:

print(f"Parent commit: {commit}")

test_files_to_run = get_diff_for_doctesting(repo, repo.head.commit, parent_commits)

all_test_files_to_run = get_all_doctest_files()

# Add to the test files to run any removed entry from "utils/not_doctested.txt".

new_test_files = get_new_doctest_files(repo, repo.head.commit, upstream_main.commit)

test_files_to_run = list(set(test_files_to_run + new_test_files))

# Do not run slow doctest tests on CircleCI

with open("utils/slow_documentation_tests.txt") as fp:

slow_documentation_tests = set(fp.read().strip().split("\n"))

test_files_to_run = [

x for x in test_files_to_run if x in all_test_files_to_run and x not in slow_documentation_tests

]

# Make sure we did not end up with a test file that was removed

test_files_to_run = [f for f in test_files_to_run if (PATH_TO_REPO / f).exists()]

"""

if cache is not None and module_fname in cache:

return cache[module_fname]

with open(PATH_TO_REPO / module_fname, "r", encoding="utf-8") as f:

content = f.read()

# Filter out all docstrings to not get imports in code examples. As before we need to deactivate formatting to

# keep this as escaped quotes and avoid this function failing on this file.

# fmt: off

splits = content.split('\"\"\"')

# fmt: on

content = "".join(splits[::2])

module_parts = str(module_fname).split(os.path.sep)

imported_modules = []

# Let's start with relative imports

relative_imports = _re_single_line_relative_imports.findall(content)

relative_imports = [

(mod, imp) for mod, imp in relative_imports if "# tests_ignore" not in imp and imp.strip() != "("

]

multiline_relative_imports = _re_multi_line_relative_imports.findall(content)

relative_imports += [(mod, imp) for mod, imp in multiline_relative_imports if "# tests_ignore" not in imp]

# We need to remove parts of the module name depending on the depth of the relative imports.

for module, imports in relative_imports:

level = 0

while module.startswith("."):

module = module[1:]

level += 1

if len(module) > 0:

dep_parts = module_parts[: len(module_parts) - level] + module.split(".")

else:

dep_parts = module_parts[: len(module_parts) - level]

imported_module = os.path.sep.join(dep_parts)

imported_modules.append((imported_module, [imp.strip() for imp in imports.split(",")]))

# Let's continue with direct imports

direct_imports = _re_single_line_direct_imports.findall(content)

direct_imports = [(mod, imp) for mod, imp in direct_imports if "# tests_ignore" not in imp and imp.strip() != "("]

multiline_direct_imports = _re_multi_line_direct_imports.findall(content)

direct_imports += [(mod, imp) for mod, imp in multiline_direct_imports if "# tests_ignore" not in imp]

# We need to find the relative path of those imports.

for module, imports in direct_imports:

import_parts = module.split(".")[1:] # ignore the name of the repo since we add it below.

dep_parts = ["src", "diffusers"] + import_parts

imported_module = os.path.sep.join(dep_parts)

imported_modules.append((imported_module, [imp.strip() for imp in imports.split(",")]))

result = []

# Double check we get proper modules (either a python file or a folder with an init).

for module_file, imports in imported_modules:

if (PATH_TO_REPO / f"{module_file}.py").is_file():

module_file = f"{module_file}.py"

elif (PATH_TO_REPO / module_file).is_dir() and (PATH_TO_REPO / module_file / "__init__.py").is_file():

module_file = os.path.sep.join([module_file, "__init__.py"])

imports = [imp for imp in imports if len(imp) > 0 and re.match("^[A-Za-z0-9_]*$", imp)]

if len(imports) > 0:

result.append((module_file, imports))

if cache is not None:

cache[module_fname] = result

"""

dependencies = []

imported_modules = extract_imports(module_fname, cache=cache)

# The while loop is to recursively traverse all inits we may encounter: we will add things as we go.

while len(imported_modules) > 0:

new_modules = []

for module, imports in imported_modules:

# If we end up in an __init__ we are often not actually importing from this init (except in the case where

# the object is fully defined in the __init__)

if module.endswith("__init__.py"):

# So we get the imports from that init then try to find where our objects come from.

new_imported_modules = extract_imports(module, cache=cache)

for new_module, new_imports in new_imported_modules:

if any(i in new_imports for i in imports):

if new_module not in dependencies:

new_modules.append((new_module, [i for i in new_imports if i in imports]))

imports = [i for i in imports if i not in new_imports]

if len(imports) > 0:

# If there are any objects lefts, they may be a submodule

path_to_module = PATH_TO_REPO / module.replace("__init__.py", "")

dependencies.extend(

[

os.path.join(module.replace("__init__.py", ""), f"{i}.py")

for i in imports

if (path_to_module / f"{i}.py").is_file()

]

)

imports = [i for i in imports if not (path_to_module / f"{i}.py").is_file()]

if len(imports) > 0:

# Then if there are still objects left, they are fully defined in the init, so we keep it as a

# dependency.

dependencies.append(module)

else:

dependencies.append(module)

imported_modules = new_modules

"""

cache = {}

all_modules = list(PATH_TO_DIFFUSERS.glob("**/*.py")) + list(PATH_TO_TESTS.glob("**/*.py"))

all_modules = [str(mod.relative_to(PATH_TO_REPO)) for mod in all_modules]

edges = [(dep, mod) for mod in all_modules for dep in get_module_dependencies(mod, cache=cache)]

"""

vertices_seen = [module]

new_edges = [edge for edge in edges if edge[0] == module and edge[1] != module and "__init__.py" not in edge[1]]

tree = [module]

while len(new_edges) > 0:

tree.append(new_edges)

final_vertices = list({edge[1] for edge in new_edges})

vertices_seen.extend(final_vertices)

new_edges = [

edge

for edge in edges

if edge[0] in final_vertices and edge[1] not in vertices_seen and "__init__.py" not in edge[1]

]

"""

if all_edges is None:

all_edges = create_reverse_dependency_tree()

tree = get_tree_starting_at(module, all_edges)

# The list of lines is a list of tuples (line_to_be_printed, module)

# Keeping the modules lets us know where to insert each new lines in the list.

lines = [(tree[0], tree[0])]

for index in range(1, len(tree)):

edges = tree[index]

start_edges = {edge[0] for edge in edges}

for start in start_edges:

end_edges = {edge[1] for edge in edges if edge[0] == start}

# We will insert all those edges just after the line showing start.

pos = 0

while lines[pos][1] != start:

pos += 1

lines = lines[: pos + 1] + [(" " * (2 * index) + end, end) for end in end_edges] + lines[pos + 1 :]

for line in lines:

# We don't print the refs that where just here to help build lines.

"""

test_example_deps = {}

all_examples = []

for framework in ["flax", "pytorch", "tensorflow"]:

test_files = list((PATH_TO_EXAMPLES / framework).glob("test_*.py"))

all_examples.extend(test_files)

# Remove the files at the root of examples/framework since they are not proper examples (they are either utils

# or example test files).

examples = [

f for f in (PATH_TO_EXAMPLES / framework).glob("**/*.py") if f.parent != PATH_TO_EXAMPLES / framework

]

all_examples.extend(examples)

for test_file in test_files:

with open(test_file, "r", encoding="utf-8") as f:

content = f.read()

# Map all examples to the test files found in examples/framework.

test_example_deps[str(test_file.relative_to(PATH_TO_REPO))] = [

str(e.relative_to(PATH_TO_REPO)) for e in examples if e.name in content

]

# Also map the test files to themselves.

test_example_deps[str(test_file.relative_to(PATH_TO_REPO))].append(

str(test_file.relative_to(PATH_TO_REPO))

)

"""

cache = {}

# Start from the example deps init.

example_deps, examples = init_test_examples_dependencies()

# Add all modules and all tests to all examples

all_modules = list(PATH_TO_DIFFUSERS.glob("**/*.py")) + list(PATH_TO_TESTS.glob("**/*.py")) + examples

all_modules = [str(mod.relative_to(PATH_TO_REPO)) for mod in all_modules]

# Compute the direct dependencies of all modules.

direct_deps = {m: get_module_dependencies(m, cache=cache) for m in all_modules}

direct_deps.update(example_deps)

# This recurses the dependencies

something_changed = True

while something_changed:

something_changed = False

for m in all_modules:

for d in direct_deps[m]:

# We stop recursing at an init (cause we always end up in the main init and we don't want to add all

# files which the main init imports)

if d.endswith("__init__.py"):

continue

if d not in direct_deps:

raise ValueError(f"KeyError:{d}. From {m}")

new_deps = set(direct_deps[d]) - set(direct_deps[m])

if len(new_deps) > 0:

direct_deps[m].extend(list(new_deps))

something_changed = True

# Finally we can build the reverse map.

reverse_map = collections.defaultdict(list)

for m in all_modules:

for d in direct_deps[m]:

reverse_map[d].append(m)

# For inits, we don't do the reverse deps but the direct deps: if modifying an init, we want to make sure we test

# all the modules impacted by that init.

for m in [f for f in all_modules if f.endswith("__init__.py")]:

direct_deps = get_module_dependencies(m, cache=cache)

deps = sum([reverse_map[d] for d in direct_deps if not d.endswith("__init__.py")], direct_deps)

reverse_map[m] = list(set(deps) - {m})

"""

if reverse_map is None:

reverse_map = create_reverse_dependency_map()

# Utility that tells us if a given file is a test (taking test examples into account)

def is_test(fname):

if fname.startswith("tests"):

return True

if fname.startswith("examples") and fname.split(os.path.sep)[-1].startswith("test"):

return True

return False

# Build the test map

test_map = {module: [f for f in deps if is_test(f)] for module, deps in reverse_map.items()}

"""

cache = {}

all_modules = list(PATH_TO_DIFFUSERS.glob("**/*.py")) + list(PATH_TO_TESTS.glob("**/*.py"))

all_modules = [str(mod.relative_to(PATH_TO_REPO)) for mod in all_modules]

direct_deps = {m: get_module_dependencies(m, cache=cache) for m in all_modules}

for module, deps in direct_deps.items():

for dep in deps:

if not (PATH_TO_REPO / dep).is_file():

"""

print(f"\n### ADD CORE PIPELINE TESTS ###\n{_print_list(IMPORTANT_PIPELINES)}")

with open(json_output_file, "rb") as fp:

test_map = json.load(fp)

# Add core pipelines as their own test group

test_map["core_pipelines"] = " ".join(

sorted([str(PATH_TO_TESTS / f"pipelines/{pipe}") for pipe in IMPORTANT_PIPELINES])

)

# If there are no existing pipeline tests save the map

if "pipelines" not in test_map:

with open(json_output_file, "w", encoding="UTF-8") as fp:

json.dump(test_map, fp, ensure_ascii=False)

pipeline_tests = test_map.pop("pipelines")

pipeline_tests = pipeline_tests.split(" ")

# Remove core pipeline tests from the fetched pipeline tests

updated_pipeline_tests = []

for pipe in pipeline_tests:

if pipe == "tests/pipelines" or Path(pipe).parts[2] in IMPORTANT_PIPELINES:

continue

updated_pipeline_tests.append(pipe)

if len(updated_pipeline_tests) > 0:

test_map["pipelines"] = " ".join(sorted(updated_pipeline_tests))

with open(json_output_file, "w", encoding="UTF-8") as fp:

"""

if json_output_file is None:

return

test_map = {}

for test_file in test_files_to_run:

# `test_file` is a path to a test folder/file, starting with `tests/`. For example,

# - `tests/models/bert/test_modeling_bert.py` or `tests/models/bert`

# - `tests/trainer/test_trainer.py` or `tests/trainer`

# - `tests/test_modeling_common.py`

names = test_file.split(os.path.sep)

module = names[1]

if module in MODULES_TO_IGNORE:

continue

if len(names) > 2 or not test_file.endswith(".py"):

# test folders under `tests` or python files under them

# take the part like tokenization, `pipeline`, etc. for other test categories

key = os.path.sep.join(names[1:2])

else:

# common test files directly under `tests/`

key = "common"

if key not in test_map:

test_map[key] = []

test_map[key].append(test_file)

# sort the keys & values

keys = sorted(test_map.keys())

test_map = {k: " ".join(sorted(test_map[k])) for k in keys}

with open(json_output_file, "w", encoding="UTF-8") as fp:

output_file: str,

diff_with_last_commit: bool = False,

json_output_file: Optional[str] = None,

):

"""

modified_files = get_modified_python_files(diff_with_last_commit=diff_with_last_commit)

print(f"\n### MODIFIED FILES ###\n{_print_list(modified_files)}")

# Create the map that will give us all impacted modules.

reverse_map = create_reverse_dependency_map()

impacted_files = modified_files.copy()

for f in modified_files:

if f in reverse_map:

impacted_files.extend(reverse_map[f])

# Remove duplicates

impacted_files = sorted(set(impacted_files))

print(f"\n### IMPACTED FILES ###\n{_print_list(impacted_files)}")

# Grab the corresponding test files:

if any(x in modified_files for x in ["setup.py"]):

test_files_to_run = ["tests", "examples"]

# in order to trigger pipeline tests even if no code change at all

if "tests/utils/tiny_model_summary.json" in modified_files:

test_files_to_run = ["tests"]

any(f.split(os.path.sep)[0] == "utils" for f in modified_files)

else:

# All modified tests need to be run.

test_files_to_run = [

f for f in modified_files if f.startswith("tests") and f.split(os.path.sep)[-1].startswith("test")

]

# Then we grab the corresponding test files.

test_map = create_module_to_test_map(reverse_map=reverse_map)

for f in modified_files:

if f in test_map:

test_files_to_run.extend(test_map[f])

test_files_to_run = sorted(set(test_files_to_run))

# Make sure we did not end up with a test file that was removed

test_files_to_run = [f for f in test_files_to_run if (PATH_TO_REPO / f).exists()]

any(f.split(os.path.sep)[0] == "utils" for f in modified_files)

examples_tests_to_run = [f for f in test_files_to_run if f.startswith("examples")]

test_files_to_run = [f for f in test_files_to_run if not f.startswith("examples")]

print(f"\n### TEST TO RUN ###\n{_print_list(test_files_to_run)}")

if len(test_files_to_run) > 0:

with open(output_file, "w", encoding="utf-8") as f:

f.write(" ".join(test_files_to_run))

# Create a map that maps test categories to test files, i.e. `models/bert` -> [...test_modeling_bert.py, ...]

# Get all test directories (and some common test files) under `tests` and `tests/models` if `test_files_to_run`

# contains `tests` (i.e. when `setup.py` is changed).

if "tests" in test_files_to_run:

test_files_to_run = get_all_tests()

create_json_map(test_files_to_run, json_output_file)

print(f"\n### EXAMPLES TEST TO RUN ###\n{_print_list(examples_tests_to_run)}")

if len(examples_tests_to_run) > 0:

# We use `all` in the case `commit_flags["test_all"]` as well as in `create_circleci_config.py` for processing

if examples_tests_to_run == ["examples"]:

examples_tests_to_run = ["all"]

example_file = Path(output_file).parent / "examples_test_list.txt"

with open(example_file, "w", encoding="utf-8") as f:

"""