Spaces:
Runtime error
Runtime error
| from __future__ import annotations | |
| import json | |
| from pathlib import Path | |
| import shutil | |
| import subprocess | |
| import apprc as rc | |
| import pytest | |
| from typer.testing import CliRunner | |
| import kneiff.cli.comfy as cli_comfy | |
| import kneiff.project_scaffold as project_scaffold_module | |
| from kneiff.config import KNEIFF_RC | |
| from kneiff.project_layout import ( | |
| DEFAULT_ANIMA_CONFIG_FILENAME, | |
| DEFAULT_FLUX2_KLEIN_9B_CONFIG_FILENAME, | |
| DEFAULT_SOURCE_DIR_NAMES, | |
| PROJECT_DEFAULT_TAGS_FILENAME, | |
| PROJECT_GITATTRIBUTES_NAME, | |
| PROJECT_GITIGNORE_NAME, | |
| PROJECT_PROMPTS_FILENAME, | |
| PROJECT_VOCABULARY_FILENAME, | |
| ) | |
| from kneiff.project_scaffold import ( | |
| DEFAULT_GIT_USER_NAME, | |
| ensure_project_resources, | |
| PROJECT_SCAFFOLD_DIR_NAMES, | |
| ) | |
| from kneiff.project_scaffold_assets import ( | |
| PROJECT_GITATTRIBUTES_RULES, | |
| PROJECT_GITIGNORE_ENTRIES, | |
| ) | |
| from kneiff.training.lora.config_templates import ( | |
| read_default_anima_starter_config, | |
| read_default_flux2_klein_9b_starter_config, | |
| ) | |
| from tests._cli_helpers import invoke_cli, set_test_config_home | |
| def test_project_init_list_use_show_and_validate( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| monkeypatch.delenv("KNF_STORAGE", raising=False) | |
| runner = CliRunner() | |
| project_root = tmp_path / "alpha" | |
| init_result = invoke_cli( | |
| runner, | |
| [ | |
| "project", | |
| "init", | |
| str(project_root), | |
| "--name", | |
| "alpha", | |
| "--activation-token", | |
| "Alpha_Character", | |
| "--species-token", | |
| "alpha_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert init_result.exit_code == 0, init_result.output | |
| assert (project_root / PROJECT_VOCABULARY_FILENAME).is_file() | |
| assert (project_root / PROJECT_PROMPTS_FILENAME).is_file() | |
| assert (project_root / PROJECT_PROMPTS_FILENAME).read_text(encoding="utf-8") == ( | |
| "version: 1\nprompts: []\n" | |
| ) | |
| assert not (project_root / "kneiff.project.yaml").exists() | |
| assert not (project_root / "resources").exists() | |
| assert (project_root / "configs").is_dir() | |
| assert (project_root / "SOURCE").is_dir() | |
| assert (project_root / "configs" / DEFAULT_ANIMA_CONFIG_FILENAME).is_file() | |
| assert (project_root / "configs" / DEFAULT_FLUX2_KLEIN_9B_CONFIG_FILENAME).is_file() | |
| for dirname in DEFAULT_SOURCE_DIR_NAMES: | |
| assert (project_root / "SOURCE" / dirname).is_dir() | |
| assert (project_root / "HF").is_dir() | |
| assert (project_root / "TRAINING").is_dir() | |
| assert not (project_root / "HF" / ".gitkeep").exists() | |
| assert not (project_root / "TRAINING" / ".gitkeep").exists() | |
| assert (project_root / PROJECT_DEFAULT_TAGS_FILENAME).read_text( | |
| encoding="utf-8" | |
| ) == "kneiff\nAlpha_Character\nalpha_species\n" | |
| assert (project_root / PROJECT_GITATTRIBUTES_NAME).is_file() | |
| assert (project_root / ".git").is_dir() | |
| branch = subprocess.run( | |
| ["git", "branch", "--show-current"], | |
| cwd=project_root, | |
| check=True, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| assert branch == "main" | |
| git_user_name = subprocess.run( | |
| ["git", "config", "--local", "--get", "user.name"], | |
| cwd=project_root, | |
| check=True, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| assert git_user_name == DEFAULT_GIT_USER_NAME | |
| assert (project_root / KNEIFF_RC.spec.storage_env_filename).read_text( | |
| encoding="utf-8" | |
| ) == "" | |
| gitignore_text = (project_root / PROJECT_GITIGNORE_NAME).read_text(encoding="utf-8") | |
| assert ".env.apprc-storage" in gitignore_text | |
| assert "HF/" in gitignore_text | |
| assert "TRAINING/" in gitignore_text | |
| for ignored_path in ("HF", "TRAINING"): | |
| ignored = subprocess.run( | |
| ["git", "check-ignore", ignored_path], | |
| cwd=project_root, | |
| check=False, | |
| capture_output=True, | |
| ) | |
| assert ignored.returncode == 0 | |
| gitattributes_rules = { | |
| line.strip() | |
| for line in (project_root / PROJECT_GITATTRIBUTES_NAME) | |
| .read_text(encoding="utf-8") | |
| .splitlines() | |
| } | |
| assert set(PROJECT_GITATTRIBUTES_RULES).issubset(gitattributes_rules) | |
| assert "Registered project: alpha" in init_result.output | |
| list_result = invoke_cli(runner, ["project", "list", "--json"]) | |
| assert list_result.exit_code == 0, list_result.output | |
| assert json.loads(list_result.output)["projects"] == [ | |
| { | |
| "error": None, | |
| "name": "alpha", | |
| "root": str(project_root.resolve()), | |
| "status": "ready", | |
| "valid": True, | |
| } | |
| ] | |
| use_result = invoke_cli(runner, ["project", "use", "alpha"]) | |
| assert use_result.exit_code == 0, use_result.output | |
| assert "persisted_default_project: alpha" in use_result.output | |
| show_result = invoke_cli(runner, ["project", "show", "--json"]) | |
| assert show_result.exit_code == 0, show_result.output | |
| payload = json.loads(show_result.output) | |
| assert payload["name"] == "alpha" | |
| assert payload["root"] == str(project_root.resolve()) | |
| assert payload["status"] == "ready" | |
| assert payload["identity"] == { | |
| "activation_token": "Alpha_Character", | |
| "species_token": "alpha_species", | |
| } | |
| assert payload["prompt_counts"]["core"] > 0 | |
| assert payload["prompt_counts"]["project"] == 0 | |
| assert payload["prompt_counts"]["resolved"] == payload["prompt_counts"]["core"] | |
| assert payload["paths"]["vocabulary"] == { | |
| "path": str(project_root / PROJECT_VOCABULARY_FILENAME), | |
| "type": "file", | |
| "status": "present", | |
| } | |
| assert payload["paths"]["storage_env"] == { | |
| "path": str(project_root / KNEIFF_RC.spec.storage_env_filename), | |
| "type": "file", | |
| "status": "present", | |
| } | |
| assert payload["paths"]["configs"]["status"] == "present" | |
| assert payload["paths"]["anima_config"] == { | |
| "path": str(project_root / "configs" / DEFAULT_ANIMA_CONFIG_FILENAME), | |
| "type": "file", | |
| "status": "present", | |
| } | |
| assert payload["paths"]["flux2_klein_9b_config"] == { | |
| "path": str(project_root / "configs" / DEFAULT_FLUX2_KLEIN_9B_CONFIG_FILENAME), | |
| "type": "file", | |
| "status": "present", | |
| } | |
| assert payload["paths"]["source"]["status"] == "present" | |
| assert payload["paths"]["source_fullbody"]["status"] == "present" | |
| assert payload["paths"]["source_head"]["status"] == "present" | |
| assert payload["paths"]["default_tags"]["status"] == "present" | |
| assert payload["paths"]["gitignore"]["status"] == "present" | |
| assert payload["paths"]["gitattributes"]["status"] == "present" | |
| assert payload["paths"]["git_repository"]["status"] == "present" | |
| assert payload["paths"]["manifest_workbook"]["status"] == "missing" | |
| assert payload["paths"]["manifest_yaml"]["status"] == "missing" | |
| assert payload["paths"]["llm_promptgen"] == { | |
| "path": str(project_root / ".llm_promptgen"), | |
| "type": "directory", | |
| "status": "missing", | |
| } | |
| assert payload["paths"]["llm_prompt_results"] == { | |
| "path": str(project_root / ".llm_promptgen" / "results"), | |
| "type": "directory", | |
| "status": "missing", | |
| } | |
| assert payload["paths"]["hf_exports"]["status"] == "present" | |
| assert payload["paths"]["training"]["status"] == "present" | |
| show_text_result = invoke_cli(runner, ["project", "show"]) | |
| assert show_text_result.exit_code == 0, show_text_result.output | |
| assert "activation_token: Alpha_Character" in show_text_result.output | |
| assert "species_token: alpha_species" in show_text_result.output | |
| assert "prompts_core:" in show_text_result.output | |
| assert "prompts_project: 0" in show_text_result.output | |
| assert "prompts_resolved:" in show_text_result.output | |
| assert f"storage_env: {project_root / KNEIFF_RC.spec.storage_env_filename}" in ( | |
| show_text_result.output | |
| ) | |
| assert f"manifest_workbook: {project_root / 'MANIFEST.knf.xlsx'}" in ( | |
| show_text_result.output | |
| ) | |
| assert f"llm_promptgen: {project_root / '.llm_promptgen'} (missing)" in ( | |
| show_text_result.output | |
| ) | |
| assert f"training: {project_root / 'TRAINING'} (present)" in ( | |
| show_text_result.output | |
| ) | |
| assert ( | |
| f"anima_config: {project_root / 'configs' / DEFAULT_ANIMA_CONFIG_FILENAME}" | |
| in (show_text_result.output) | |
| ) | |
| assert ( | |
| "flux2_klein_9b_config: " | |
| f"{project_root / 'configs' / DEFAULT_FLUX2_KLEIN_9B_CONFIG_FILENAME}" | |
| in show_text_result.output | |
| ) | |
| validate_result = invoke_cli(runner, ["project", "validate"]) | |
| assert validate_result.exit_code == 0, validate_result.output | |
| assert "activation_token: Alpha_Character" in validate_result.output | |
| assert "species_token: alpha_species" in validate_result.output | |
| assert "prompts_core:" in validate_result.output | |
| assert "prompts_project: 0" in validate_result.output | |
| assert "prompts_resolved:" in validate_result.output | |
| assert f"anima_config: {DEFAULT_ANIMA_CONFIG_FILENAME}" in validate_result.output | |
| assert ( | |
| f"flux2_klein_9b_config: {DEFAULT_FLUX2_KLEIN_9B_CONFIG_FILENAME}" | |
| in validate_result.output | |
| ) | |
| assert "starter_source_directories: 0-FULLBODY, 2-HEAD" in validate_result.output | |
| (project_root / PROJECT_PROMPTS_FILENAME).unlink() | |
| optional_show_result = invoke_cli(runner, ["project", "show", "--json"]) | |
| optional_list_result = invoke_cli(runner, ["project", "list", "--json"]) | |
| optional_validate_result = invoke_cli(runner, ["project", "validate"]) | |
| assert optional_show_result.exit_code == 0, optional_show_result.output | |
| optional_payload = json.loads(optional_show_result.output) | |
| assert optional_payload["status"] == "ready" | |
| assert optional_payload["paths"]["prompts"]["status"] == "missing" | |
| assert optional_payload["prompt_counts"]["project"] == 0 | |
| assert optional_list_result.exit_code == 0, optional_list_result.output | |
| assert json.loads(optional_list_result.output)["projects"][0]["valid"] is True | |
| assert optional_validate_result.exit_code == 0, optional_validate_result.output | |
| assert "prompts_project: 0" in optional_validate_result.output | |
| def test_project_init_requires_identity_for_a_new_vocabulary( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| runner = CliRunner() | |
| result = invoke_cli( | |
| runner, | |
| ["project", "init", str(tmp_path / "missing-identity"), "--yes"], | |
| ) | |
| assert result.exit_code == 1 | |
| assert "requires both an activation token and a species token" in result.output | |
| assert not (tmp_path / "missing-identity").exists() | |
| def test_project_show_and_validate_report_missing_project_resources( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "plain-storage" | |
| root.mkdir() | |
| for dirname in PROJECT_SCAFFOLD_DIR_NAMES: | |
| (root / dirname).mkdir() | |
| rc.storage.register_storage( | |
| name="plain", | |
| root=root, | |
| path=rc.storage.index_path_for_create(KNEIFF_RC.spec), | |
| storage_env_filename=KNEIFF_RC.spec.storage_env_filename, | |
| ) | |
| monkeypatch.setenv("KNF_STORAGE", "plain") | |
| result = invoke_cli(CliRunner(), ["project", "show", "--json"]) | |
| assert result.exit_code == 0, result.output | |
| payload = json.loads(result.output) | |
| assert payload["paths"]["vocabulary"]["status"] == "missing" | |
| assert payload["paths"]["prompts"]["status"] == "missing" | |
| list_result = invoke_cli(CliRunner(), ["project", "list", "--json"]) | |
| assert list_result.exit_code == 0, list_result.output | |
| assert json.loads(list_result.output)["projects"][0]["valid"] is False | |
| validate_result = invoke_cli(CliRunner(), ["project", "validate"]) | |
| assert validate_result.exit_code == 2 | |
| assert PROJECT_VOCABULARY_FILENAME in "".join(validate_result.output.split()) | |
| def test_project_commands_report_missing_required_scaffold_directory( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| dirname: str, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| runner = CliRunner() | |
| project_root = tmp_path / "incomplete" | |
| init_result = invoke_cli( | |
| runner, | |
| [ | |
| "project", | |
| "init", | |
| str(project_root), | |
| "--name", | |
| "incomplete", | |
| "--activation-token", | |
| "Incomplete_Character", | |
| "--species-token", | |
| "incomplete_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert init_result.exit_code == 0, init_result.output | |
| shutil.rmtree(project_root / dirname) | |
| monkeypatch.setenv("KNF_STORAGE", "incomplete") | |
| show_result = invoke_cli(runner, ["project", "show", "--json"]) | |
| list_result = invoke_cli(runner, ["project", "list", "--json"]) | |
| validate_result = invoke_cli(runner, ["project", "validate"]) | |
| assert show_result.exit_code == 0, show_result.output | |
| show_payload = json.loads(show_result.output) | |
| assert show_payload["status"] == "invalid" | |
| assert dirname in show_payload["error"] | |
| assert list_result.exit_code == 0, list_result.output | |
| [list_payload] = json.loads(list_result.output)["projects"] | |
| assert list_payload["valid"] is False | |
| assert list_payload["status"] == "invalid" | |
| assert dirname in list_payload["error"] | |
| assert validate_result.exit_code == 2 | |
| validation_output = "".join(validate_result.output.replace("│", "").split()) | |
| assert dirname in validation_output | |
| def test_project_registry_rejects_symlinked_registered_root( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| project_root = tmp_path / "real-project" | |
| ensure_project_resources( | |
| project_root, | |
| activation_token="Demo_Character", | |
| species_token="demo_species", | |
| ) | |
| registered_root = tmp_path / "registered-project" | |
| registered_root.symlink_to(project_root, target_is_directory=True) | |
| registry_path = rc.storage.index_path_for_create(KNEIFF_RC.spec) | |
| registry_path.parent.mkdir(parents=True, exist_ok=True) | |
| rc.storage.write_storage_registry( | |
| rc.storage.StorageRegistry( | |
| path=registry_path, | |
| storages={ | |
| "linked": rc.storage.StorageRecord( | |
| name="linked", | |
| root=registered_root, | |
| ) | |
| }, | |
| ) | |
| ) | |
| list_result = invoke_cli(CliRunner(), ["project", "list", "--json"]) | |
| use_result = invoke_cli(CliRunner(), ["project", "use", "linked"]) | |
| assert list_result.exit_code == 0, list_result.output | |
| [row] = json.loads(list_result.output)["projects"] | |
| assert row["valid"] is False | |
| assert "Project root must not be a symlink" in row["error"] | |
| assert use_result.exit_code == 2 | |
| assert "Cannot select project 'linked'" in use_result.output | |
| def test_project_init_preserves_existing_repository_git_identity( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "adopted" | |
| root.mkdir() | |
| subprocess.run(["git", "init"], cwd=root, check=True, capture_output=True) | |
| subprocess.run( | |
| ["git", "config", "--local", "user.name", "Existing Author"], | |
| cwd=root, | |
| check=True, | |
| ) | |
| subprocess.run( | |
| [ | |
| "git", | |
| "config", | |
| "--local", | |
| "user.email", | |
| "kneiff@users.noreply.huggingface.co", | |
| ], | |
| cwd=root, | |
| check=True, | |
| ) | |
| result = invoke_cli( | |
| CliRunner(), | |
| [ | |
| "project", | |
| "init", | |
| str(root), | |
| "--activation-token", | |
| "Adopted_Character", | |
| "--species-token", | |
| "adopted_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert result.exit_code == 0, result.output | |
| name = subprocess.run( | |
| ["git", "config", "--local", "--get", "user.name"], | |
| cwd=root, | |
| check=True, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| email = subprocess.run( | |
| ["git", "config", "--local", "--get", "user.email"], | |
| cwd=root, | |
| check=True, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| assert (name, email) == ( | |
| "Existing Author", | |
| "kneiff@users.noreply.huggingface.co", | |
| ) | |
| def test_project_init_sets_requested_local_git_identity_without_initial_commit( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "custom-git-user" | |
| result = invoke_cli( | |
| CliRunner(), | |
| [ | |
| "project", | |
| "init", | |
| str(root), | |
| "--activation-token", | |
| "Demo_Character", | |
| "--species-token", | |
| "demo_species", | |
| "--git-user-name", | |
| "Kneiff Starter Author", | |
| "--yes", | |
| ], | |
| ) | |
| assert result.exit_code == 0, result.output | |
| user_name = subprocess.run( | |
| ["git", "config", "--local", "--get", "user.name"], | |
| cwd=root, | |
| check=True, | |
| capture_output=True, | |
| text=True, | |
| ).stdout.strip() | |
| no_local_email = subprocess.run( | |
| ["git", "config", "--local", "--get", "user.email"], | |
| cwd=root, | |
| check=False, | |
| capture_output=True, | |
| ) | |
| no_initial_commit = subprocess.run( | |
| ["git", "rev-parse", "--verify", "HEAD"], | |
| cwd=root, | |
| check=False, | |
| capture_output=True, | |
| ) | |
| remotes = subprocess.run( | |
| ["git", "remote"], | |
| cwd=root, | |
| check=True, | |
| capture_output=True, | |
| text=True, | |
| ).stdout | |
| assert user_name == "Kneiff Starter Author" | |
| assert no_local_email.returncode != 0 | |
| assert no_initial_commit.returncode != 0 | |
| assert remotes == "" | |
| def test_project_init_adopts_project_owned_files_without_replacing_them( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "adopted-content" | |
| root.mkdir() | |
| subprocess.run( | |
| ["git", "init", "--initial-branch", "main"], | |
| cwd=root, | |
| check=True, | |
| capture_output=True, | |
| ) | |
| subprocess.run( | |
| ["git", "config", "--local", "user.name", "Existing Author"], | |
| cwd=root, | |
| check=True, | |
| ) | |
| subprocess.run( | |
| [ | |
| "git", | |
| "config", | |
| "--local", | |
| "user.email", | |
| "kneiff@users.noreply.huggingface.co", | |
| ], | |
| cwd=root, | |
| check=True, | |
| ) | |
| ensure_project_resources( | |
| root, | |
| activation_token="Adopted_Character", | |
| species_token="adopted_species", | |
| ) | |
| configs_path = root / "configs" | |
| configs_path.mkdir() | |
| anima_config_path = configs_path / DEFAULT_ANIMA_CONFIG_FILENAME | |
| anima_config_path.write_text( | |
| "# Existing project config.\n" + read_default_anima_starter_config(), | |
| encoding="utf-8", | |
| ) | |
| flux2_klein_9b_config_path = configs_path / DEFAULT_FLUX2_KLEIN_9B_CONFIG_FILENAME | |
| flux2_klein_9b_config_path.write_text( | |
| "# Existing Flux2 project config.\n" | |
| + read_default_flux2_klein_9b_starter_config(), | |
| encoding="utf-8", | |
| ) | |
| default_tags_path = root / PROJECT_DEFAULT_TAGS_FILENAME | |
| default_tags_path.write_text("existing tags\n", encoding="utf-8") | |
| git_config_path = root / ".git" / "config" | |
| expected_bytes = { | |
| path: path.read_bytes() | |
| for path in ( | |
| root / PROJECT_VOCABULARY_FILENAME, | |
| root / PROJECT_PROMPTS_FILENAME, | |
| anima_config_path, | |
| flux2_klein_9b_config_path, | |
| default_tags_path, | |
| git_config_path, | |
| ) | |
| } | |
| result = invoke_cli( | |
| CliRunner(), | |
| [ | |
| "project", | |
| "init", | |
| str(root), | |
| "--git-user-name", | |
| "Must Not Replace Existing Author", | |
| "--yes", | |
| ], | |
| ) | |
| assert result.exit_code == 0, result.output | |
| assert {path: path.read_bytes() for path in expected_bytes} == expected_bytes | |
| for dirname in DEFAULT_SOURCE_DIR_NAMES: | |
| assert (root / "SOURCE" / dirname).is_dir() | |
| assert (root / "SOURCE" / "2-HEAD").is_dir() | |
| def test_project_selection_precedence( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| monkeypatch.delenv("KNF_STORAGE", raising=False) | |
| runner = CliRunner() | |
| alpha_root = tmp_path / "alpha" | |
| beta_root = tmp_path / "beta" | |
| for name, root in (("alpha", alpha_root), ("beta", beta_root)): | |
| result = invoke_cli( | |
| runner, | |
| [ | |
| "project", | |
| "init", | |
| str(root), | |
| "--name", | |
| name, | |
| "--activation-token", | |
| f"{name}_Character", | |
| "--species-token", | |
| f"{name}_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert result.exit_code == 0, result.output | |
| use_result = invoke_cli(runner, ["project", "use", "alpha"]) | |
| assert use_result.exit_code == 0, use_result.output | |
| monkeypatch.setenv("KNF_STORAGE", "beta") | |
| root_override = invoke_cli( | |
| runner, | |
| ["--storage", "alpha", "project", "show", "--json"], | |
| ) | |
| assert root_override.exit_code == 0, root_override.output | |
| assert json.loads(root_override.output)["root"] == str(alpha_root.resolve()) | |
| shell_override = invoke_cli(runner, ["project", "show", "--json"]) | |
| assert shell_override.exit_code == 0, shell_override.output | |
| assert json.loads(shell_override.output)["root"] == str(beta_root.resolve()) | |
| def test_project_init_rejects_invalid_name_before_writing( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "invalid-name" | |
| result = invoke_cli( | |
| CliRunner(), | |
| [ | |
| "project", | |
| "init", | |
| str(root), | |
| "--name", | |
| "bad name", | |
| "--activation-token", | |
| "Demo_Character", | |
| "--species-token", | |
| "demo_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert result.exit_code == 1 | |
| assert "Project names may contain only" in result.output | |
| assert not root.exists() | |
| def test_project_init_rejects_selector_and_root_rebinding( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| runner = CliRunner() | |
| alpha = tmp_path / "alpha" | |
| beta = tmp_path / "beta" | |
| first = invoke_cli( | |
| runner, | |
| [ | |
| "project", | |
| "init", | |
| str(alpha), | |
| "--name", | |
| "alpha", | |
| "--activation-token", | |
| "Alpha_Character", | |
| "--species-token", | |
| "alpha_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert first.exit_code == 0, first.output | |
| name_collision = invoke_cli( | |
| runner, | |
| [ | |
| "project", | |
| "init", | |
| str(beta), | |
| "--name", | |
| "alpha", | |
| "--activation-token", | |
| "Beta_Character", | |
| "--species-token", | |
| "beta_species", | |
| "--yes", | |
| ], | |
| ) | |
| root_collision = invoke_cli( | |
| runner, | |
| ["project", "init", str(alpha), "--name", "alias", "--yes"], | |
| ) | |
| assert name_collision.exit_code == 1 | |
| assert "refusing to rebind" in name_collision.output | |
| assert not beta.exists() | |
| assert root_collision.exit_code == 1 | |
| assert "already registered as 'alpha'" in root_collision.output | |
| def test_project_init_rejects_invalid_existing_prompts_without_mutation( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "invalid-prompts" | |
| root.mkdir() | |
| (root / PROJECT_VOCABULARY_FILENAME).write_text( | |
| """ | |
| version: 1 | |
| custom_tokens: | |
| character: | |
| Demo_Character: | |
| text: Demo_Character | |
| species: | |
| demo_species: | |
| text: demo_species | |
| axes: [] | |
| """, | |
| encoding="utf-8", | |
| ) | |
| (root / PROJECT_PROMPTS_FILENAME).mkdir() | |
| before = sorted(path.name for path in root.iterdir()) | |
| result = invoke_cli( | |
| CliRunner(), | |
| ["project", "init", str(root), "--name", "broken", "--yes"], | |
| ) | |
| assert result.exit_code == 1 | |
| assert "exists and is not a file" in result.output | |
| assert sorted(path.name for path in root.iterdir()) == before | |
| registry = rc.storage.load_storage_registry_or_empty( | |
| rc.storage.index_path_for_create(KNEIFF_RC.spec) | |
| ) | |
| assert "broken" not in registry.storages | |
| def test_project_init_rejects_resource_symlinks_without_following_them( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "linked" | |
| root.mkdir() | |
| external_vocabulary = tmp_path / "external-vocabulary.yaml" | |
| (root / PROJECT_VOCABULARY_FILENAME).symlink_to(external_vocabulary) | |
| result = invoke_cli( | |
| CliRunner(), | |
| [ | |
| "project", | |
| "init", | |
| str(root), | |
| "--activation-token", | |
| "Demo_Character", | |
| "--species-token", | |
| "demo_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert result.exit_code == 1 | |
| assert "must not be a symlink" in result.output | |
| assert not external_vocabulary.exists() | |
| def test_project_init_rejects_invalid_storage_environment_before_registration( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| invalid_kind: str, | |
| error: str, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "invalid-storage-env" | |
| root.mkdir() | |
| storage_env_path = root / KNEIFF_RC.spec.storage_env_filename | |
| external_env = tmp_path / "external.env" | |
| if invalid_kind == "symlink": | |
| external_env.write_text("KNF_WORKERS=7\n", encoding="utf-8") | |
| storage_env_path.symlink_to(external_env) | |
| else: | |
| storage_env_path.mkdir() | |
| registration_called = False | |
| def unexpected_registration(**_kwargs: object) -> None: | |
| nonlocal registration_called | |
| registration_called = True | |
| monkeypatch.setattr(rc.storage, "register_storage", unexpected_registration) | |
| result = invoke_cli( | |
| CliRunner(), | |
| [ | |
| "project", | |
| "init", | |
| str(root), | |
| "--name", | |
| "invalid-storage-env", | |
| "--activation-token", | |
| "Demo_Character", | |
| "--species-token", | |
| "demo_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert result.exit_code == 1 | |
| assert error in result.output | |
| assert registration_called is False | |
| if invalid_kind == "symlink": | |
| assert external_env.read_text(encoding="utf-8") == "KNF_WORKERS=7\n" | |
| def test_project_init_rejects_gitignore_symlink_without_rewriting_target( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "linked-ignore" | |
| ensure_project_resources( | |
| root, | |
| activation_token="Demo_Character", | |
| species_token="demo_species", | |
| ) | |
| external_gitignore = tmp_path / "external.gitignore" | |
| external_gitignore.write_text("keep-this\n", encoding="utf-8") | |
| (root / ".gitignore").symlink_to(external_gitignore) | |
| result = invoke_cli( | |
| CliRunner(), | |
| ["project", "init", str(root), "--yes"], | |
| ) | |
| assert result.exit_code == 1 | |
| assert "must not be a symlink" in result.output | |
| assert external_gitignore.read_text(encoding="utf-8") == "keep-this\n" | |
| def test_project_init_gitignore_covers_generated_project_artifacts( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "ignored" | |
| result = invoke_cli( | |
| CliRunner(), | |
| [ | |
| "project", | |
| "init", | |
| str(root), | |
| "--activation-token", | |
| "Demo_Character", | |
| "--species-token", | |
| "demo_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert result.exit_code == 0, result.output | |
| entries = set((root / ".gitignore").read_text(encoding="utf-8").splitlines()) | |
| assert set(PROJECT_GITIGNORE_ENTRIES).issubset(entries) | |
| def test_project_init_rolls_back_new_scaffold_when_registration_fails( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "new-parent" / "new-project" | |
| def fail_registration(**_kwargs: object) -> None: | |
| raise OSError("registry write failed") | |
| monkeypatch.setattr(rc.storage, "register_storage", fail_registration) | |
| result = invoke_cli( | |
| CliRunner(), | |
| [ | |
| "project", | |
| "init", | |
| str(root), | |
| "--name", | |
| "new-project", | |
| "--activation-token", | |
| "Demo_Character", | |
| "--species-token", | |
| "demo_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert result.exit_code == 1 | |
| assert "AppRC registration failed" in result.output | |
| assert not root.parent.exists() | |
| def test_project_init_rolls_back_git_metadata_when_local_identity_setup_fails( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "new-parent" / "new-project" | |
| real_run = subprocess.run | |
| def fail_git_config( | |
| arguments: tuple[str, ...], | |
| *, | |
| check: bool, | |
| capture_output: bool, | |
| text: bool, | |
| ) -> subprocess.CompletedProcess[str]: | |
| if "config" in arguments: | |
| return subprocess.CompletedProcess( | |
| arguments, | |
| returncode=1, | |
| stdout="", | |
| stderr="forced local identity failure", | |
| ) | |
| return real_run( | |
| arguments, | |
| check=check, | |
| capture_output=capture_output, | |
| text=text, | |
| ) | |
| monkeypatch.setattr(project_scaffold_module.subprocess, "run", fail_git_config) | |
| result = invoke_cli( | |
| CliRunner(), | |
| [ | |
| "project", | |
| "init", | |
| str(root), | |
| "--activation-token", | |
| "Demo_Character", | |
| "--species-token", | |
| "demo_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert result.exit_code == 1 | |
| assert "forced local identity failure" in result.output | |
| assert not root.parent.exists() | |
| def test_project_init_restores_adopted_root_when_registration_fails( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| root = tmp_path / "adopted" | |
| root.mkdir() | |
| sentinel = root / "keep.txt" | |
| sentinel.write_text("keep\n", encoding="utf-8") | |
| gitignore_path = root / ".gitignore" | |
| gitignore_path.write_bytes(b"custom-entry\n") | |
| gitattributes_path = root / ".gitattributes" | |
| gitattributes_path.write_bytes(b"*.custom filter=custom\n") | |
| def fail_registration(**_kwargs: object) -> None: | |
| raise OSError("registry write failed") | |
| monkeypatch.setattr(rc.storage, "register_storage", fail_registration) | |
| result = invoke_cli( | |
| CliRunner(), | |
| [ | |
| "project", | |
| "init", | |
| str(root), | |
| "--name", | |
| "adopted", | |
| "--activation-token", | |
| "Demo_Character", | |
| "--species-token", | |
| "demo_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert result.exit_code == 1 | |
| assert sentinel.read_text(encoding="utf-8") == "keep\n" | |
| assert gitignore_path.read_bytes() == b"custom-entry\n" | |
| assert gitattributes_path.read_bytes() == b"*.custom filter=custom\n" | |
| assert sorted(path.name for path in root.iterdir()) == [ | |
| ".gitattributes", | |
| ".gitignore", | |
| "keep.txt", | |
| ] | |
| def test_project_registry_commands_report_errors_without_tracebacks( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| missing = invoke_cli(CliRunner(), ["project", "use", "missing"]) | |
| assert missing.exit_code == 2 | |
| assert "Cannot select project 'missing'" in missing.output | |
| assert "Traceback" not in missing.output | |
| registry_path = rc.storage.index_path_for_create(KNEIFF_RC.spec) | |
| registry_path.parent.mkdir(parents=True, exist_ok=True) | |
| registry_path.write_text("not = [valid", encoding="utf-8") | |
| malformed = invoke_cli(CliRunner(), ["project", "list"]) | |
| assert malformed.exit_code == 2 | |
| assert "Cannot load project registry" in malformed.output | |
| assert "Traceback" not in malformed.output | |
| def test_project_use_rejects_invalid_registered_project( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| runner = CliRunner() | |
| project_root = tmp_path / "broken" | |
| init_result = invoke_cli( | |
| runner, | |
| [ | |
| "project", | |
| "init", | |
| str(project_root), | |
| "--name", | |
| "broken", | |
| "--activation-token", | |
| "Broken_Character", | |
| "--species-token", | |
| "broken_species", | |
| "--yes", | |
| ], | |
| ) | |
| assert init_result.exit_code == 0, init_result.output | |
| (project_root / PROJECT_PROMPTS_FILENAME).write_text( | |
| "version: 1\nprompts: not-a-list\n", | |
| encoding="utf-8", | |
| ) | |
| result = invoke_cli(runner, ["project", "use", "broken"]) | |
| assert result.exit_code == 2 | |
| assert "Cannot select project 'broken'" in result.output | |
| normalized_output = " ".join(result.output.replace("│", "").split()) | |
| assert "Prompt catalog must contain a prompts list." in normalized_output | |
| def test_project_list_rejects_ignored_runtime_options( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| root_options: list[str], | |
| ) -> None: | |
| set_test_config_home(monkeypatch, tmp_path) | |
| result = invoke_cli(CliRunner(), [*root_options, "project", "list"]) | |
| assert result.exit_code == 2 | |
| assert "do not apply to" in result.output | |
| assert "runtime-independent" in result.output | |
| assert "project list" in result.output | |
| def test_comfy_upscale_does_not_require_project_resources( | |
| monkeypatch: pytest.MonkeyPatch, | |
| tmp_path: Path, | |
| isolated_cli_storage: Path, | |
| ) -> None: | |
| import kneiff.infer.comfy.upscale as upscale_module | |
| class FakeModelClient: | |
| def __init__(self, _server_url: str) -> None: | |
| pass | |
| def input_options(self, _node_class: str, _input_name: str) -> tuple[str, ...]: | |
| return ("local-upscaler.safetensors",) | |
| monkeypatch.setattr(cli_comfy, "ComfyUiClient", FakeModelClient) | |
| monkeypatch.setattr(upscale_module, "run_upscale", lambda *_args, **_kwargs: ()) | |
| image_path = tmp_path / "source.png" | |
| image_path.write_bytes(b"image") | |
| result = invoke_cli( | |
| CliRunner(), | |
| [ | |
| "comfy", | |
| "upscale", | |
| str(image_path), | |
| "--fast", | |
| "--model", | |
| "local-upscaler.safetensors", | |
| ], | |
| ) | |
| assert result.exit_code == 0, result.output | |
| assert not (isolated_cli_storage / PROJECT_VOCABULARY_FILENAME).exists() | |
| assert not (isolated_cli_storage / PROJECT_PROMPTS_FILENAME).exists() | |