Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ dependencies = [
"pytest-cov",
"pytest-timeout",
"pytest-datadir",
"requests_testadapter",
"requests_mock",
"httpx",
"polyfactory==2.19.0",
"humanize==4.12.1",
Expand Down
4 changes: 0 additions & 4 deletions tests/app/services/download/test_download_queue.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@
import pytest
from pydantic.networks import AnyHttpUrl
from requests.sessions import Session
from requests_testadapter import TestAdapter

from invokeai.app.services.config import get_config
from invokeai.app.services.config.config_default import URLRegexTokenPair
Expand All @@ -24,9 +23,6 @@
from invokeai.backend.model_manager.metadata import HuggingFaceMetadataFetch, ModelMetadataWithFiles, RemoteModelFile
from tests.test_nodes import TestEventService

# Prevent pytest deprecation warnings
TestAdapter.__test__ = False


@pytest.mark.timeout(timeout=10, method="thread")
def test_basic_queue_download(tmp_path: Path, mm2_session: Session) -> None:
Expand Down
178 changes: 91 additions & 87 deletions tests/backend/model_manager/model_manager_fixtures.py
Original file line number Diff line number Diff line change
@@ -1,40 +1,40 @@
# Fixtures to support testing of the model_manager v2 installer, metadata and record store

import os
import shutil
from pathlib import Path

import pytest
from requests.sessions import Session
from requests_testadapter import TestAdapter, TestSession
import requests_mock

from invokeai.app.services.config import InvokeAIAppConfig
from invokeai.app.services.download import DownloadQueueService, DownloadQueueServiceBase
from invokeai.app.services.model_install import ModelInstallService, ModelInstallServiceBase
from invokeai.app.services.model_load import ModelLoadService, ModelLoadServiceBase
from invokeai.app.services.model_manager import ModelManagerService, ModelManagerServiceBase
from invokeai.app.services.model_records import ModelRecordServiceBase, ModelRecordServiceSQL
from invokeai.backend.model_manager import BaseModelType, ModelFormat, ModelType, ModelVariantType
from invokeai.backend.model_manager.config import (
LoRADiffusersConfig,
MainCheckpointConfig,
MainDiffusersConfig,
VAEDiffusersConfig,
)
from invokeai.backend.model_manager.load.model_cache.model_cache import ModelCache
from invokeai.backend.model_manager.taxonomy import ModelSourceType
from invokeai.backend.util.devices import TorchDevice
from invokeai.backend.util.logging import InvokeAILogger
from tests.backend.model_manager.model_metadata.metadata_examples import (
HFTestLoraMetadata,
RepoCivitaiModelMetadata1,
RepoCivitaiVersionMetadata1,
RepoHFMetadata1,
RepoHFMetadata1_nofp16,
RepoHFModelJson1,
)
from tests.fixtures.sqlite_database import create_mock_sqlite_database
from tests.test_nodes import TestEventService

Check failure on line 37 in tests/backend/model_manager/model_manager_fixtures.py

View workflow job for this annotation

GitHub Actions / python-checks

Ruff (I001)

tests/backend/model_manager/model_manager_fixtures.py:3:1: I001 Import block is un-sorted or un-formatted


# Create a temporary directory using the contents of `./data/invokeai_root` as the template
Expand Down Expand Up @@ -213,81 +213,83 @@
@pytest.fixture
def mm2_session(embedding_file: Path, diffusers_dir: Path) -> Session:
"""This fixtures defines a series of mock URLs for testing download and installation."""
sess: Session = TestSession()
sess.mount(
sess = Session()
adapter = requests_mock.Adapter()
sess.mount("http://", adapter)
sess.mount("https://", adapter)
adapter.register_uri(
"GET",
"https://test.com/missing_model.safetensors",
TestAdapter(
b"missing",
status=404,
),
text="missing",
status_code=404,
reason="NOT FOUND",
)
sess.mount(
adapter.register_uri(
"GET",
"https://huggingface.co/api/models/stabilityai/sdxl-turbo",
TestAdapter(
RepoHFMetadata1,
headers={"Content-Type": "application/json; charset=utf-8", "Content-Length": len(RepoHFMetadata1)},
),
)
sess.mount(
"https://huggingface.co/api/models/stabilityai/sdxl-turbo-nofp16",
TestAdapter(
RepoHFMetadata1_nofp16,
headers={"Content-Type": "application/json; charset=utf-8", "Content-Length": len(RepoHFMetadata1_nofp16)},
),
content=RepoHFMetadata1,
headers={"Content-Type": "application/json; charset=utf-8", "Content-Length": str(len(RepoHFMetadata1))},
)
sess.mount(
"https://civitai.com/api/v1/model-versions/242807",
TestAdapter(
RepoCivitaiVersionMetadata1,
(
adapter.register_uri(
"GET",
"https://huggingface.co/api/models/stabilityai/sdxl-turbo-nofp16",
content=RepoHFMetadata1_nofp16,
headers={
"Content-Length": len(RepoCivitaiVersionMetadata1),
"Content-Type": "application/json; charset=utf-8",
"Content-Length": str(len(RepoHFMetadata1_nofp16)),
},
),
)
sess.mount(
adapter.register_uri(
"GET",
"https://civitai.com/api/v1/model-versions/242807",
content=RepoCivitaiVersionMetadata1,
headers={"Content-Length": str(len(RepoCivitaiVersionMetadata1))},
)
adapter.register_uri(
"GET",
"https://civitai.com/api/v1/models/215485",
TestAdapter(
RepoCivitaiModelMetadata1,
headers={
"Content-Length": len(RepoCivitaiModelMetadata1),
},
),
content=RepoCivitaiModelMetadata1,
headers={"Content-Length": str(len(RepoCivitaiModelMetadata1))},
)
sess.mount(
adapter.register_uri(
"GET",
"https://huggingface.co/stabilityai/sdxl-turbo/resolve/main/model_index.json",
TestAdapter(
RepoHFModelJson1,
headers={
"Content-Length": len(RepoHFModelJson1),
},
),
content=RepoHFModelJson1,
headers={"Content-Length": str(len(RepoHFModelJson1))},
)
with open(embedding_file, "rb") as f:
data = f.read() # file is small - just 15K
sess.mount(
adapter.register_uri(
"GET",
"https://www.test.foo/download/test_embedding.safetensors",
TestAdapter(data, headers={"Content-Type": "application/octet-stream", "Content-Length": len(data)}),
content=data,
headers={"Content-Type": "application/octet-stream", "Content-Length": str(len(data))},
)
sess.mount(
adapter.register_uri(
"GET",
"https://huggingface.co/api/models/stabilityai/sdxl-turbo",
TestAdapter(
RepoHFMetadata1,
headers={"Content-Type": "application/json; charset=utf-8", "Content-Length": len(RepoHFMetadata1)},
),
content=RepoHFMetadata1,
headers={"Content-Type": "application/json; charset=utf-8", "Content-Length": str(len(RepoHFMetadata1))},
)
adapter.register_uri(
"GET",
"https://huggingface.co/api/models/stabilityai/sdxl-turbo/revision/ModelRepoVariant.Default?blobs=True",
content=RepoHFMetadata1,
headers={"Content-Type": "application/json; charset=utf-8", "Content-Length": str(len(RepoHFMetadata1))},
)
sess.mount(
adapter.register_uri(
"GET",
"https://huggingface.co/api/models/InvokeAI-test/textual_inversion_tests?blobs=True",
TestAdapter(
HFTestLoraMetadata,
headers={"Content-Type": "application/json; charset=utf-8", "Content-Length": len(HFTestLoraMetadata)},
),
content=HFTestLoraMetadata,
headers={"Content-Type": "application/json; charset=utf-8", "Content-Length": str(len(HFTestLoraMetadata))},
)
sess.mount(
adapter.register_uri(
"GET",
"https://huggingface.co/InvokeAI-test/textual_inversion_tests/resolve/main/learned_embeds-steps-1000.safetensors",
TestAdapter(
data,
headers={"Content-Type": "application/json; charset=utf-8", "Content-Length": len(data)},
),
content=data,
headers={"Content-Type": "application/json; charset=utf-8", "Content-Length": str(len(data))},
)
for root, _, files in os.walk(diffusers_dir):
for name in files:
Expand All @@ -296,55 +298,57 @@
url = f"https://huggingface.co/stabilityai/sdxl-turbo/resolve/main/{url_base}"
with open(path, "rb") as f:
data = f.read()
sess.mount(
adapter.register_uri(
"GET",
url,
TestAdapter(
data,
headers={
"Content-Type": "application/json; charset=utf-8",
"Content-Length": len(data),
},
),
content=data,
headers={
"Content-Type": "application/json; charset=utf-8",
"Content-Length": str(len(data)),
},
)

for i in ["12345", "9999", "54321"]:
content = (
b"I am a safetensors file " + bytearray(i, "utf-8") + bytearray(32_000)
) # for pause tests, must make content large
sess.mount(
adapter.register_uri(
"GET",
f"http://www.civitai.com/models/{i}",
TestAdapter(
content,
headers={
"Content-Length": len(content),
"Content-Disposition": f'filename="mock{i}.safetensors"',
},
),
content=content,
headers={
"Content-Length": str(len(content)),
"Content-Disposition": f'filename="mock{i}.safetensors"',
},
)

sess.mount(
adapter.register_uri(
"GET",
"http://www.huggingface.co/foo.txt",
TestAdapter(
content,
headers={
"Content-Length": len(content),
"Content-Disposition": 'filename="foo.safetensors"',
},
),
content=content,
headers={
"Content-Length": str(len(content)),
"Content-Disposition": 'filename="foo.safetensors"',
},
)

# here are some malformed URLs to test
# missing the content length
sess.mount(
adapter.register_uri(
"GET",
"http://www.civitai.com/models/missing",
TestAdapter(
b"Missing content length",
headers={
"Content-Disposition": 'filename="missing.txt"',
},
),
text="Missing content length",
headers={
"Content-Disposition": 'filename="missing.txt"',
},
)
# not found test
sess.mount("http://www.civitai.com/models/broken", TestAdapter(b"Not found", status=404))
adapter.register_uri(
"GET",
"http://www.civitai.com/models/broken",
text="Not found",
status_code=404,
reason="NOT FOUND",
)

return sess
Loading