mirror of
https://github.com/comfyanonymous/ComfyUI.git
synced 2026-05-30 10:57:23 +08:00
Compare commits
3 Commits
f350a175c3
...
2001646f78
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2001646f78 | ||
|
|
22186b3dae | ||
|
|
06e416bd0d |
@ -1,3 +1,4 @@
|
||||
import functools
|
||||
import logging
|
||||
import os
|
||||
|
||||
@ -5,29 +6,24 @@ import folder_paths
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_DEFAULT_DEPLOY_ENV = "local_git"
|
||||
_DEFAULT_DEPLOY_ENV = "local-git"
|
||||
_ENV_FILENAME = ".comfy_environment"
|
||||
|
||||
_cached_value: str | None = None
|
||||
|
||||
|
||||
@functools.cache
|
||||
def get_deploy_environment() -> str:
|
||||
global _cached_value
|
||||
if _cached_value is not None:
|
||||
return _cached_value
|
||||
|
||||
env_file = os.path.join(folder_paths.base_path, _ENV_FILENAME)
|
||||
try:
|
||||
with open(env_file, encoding="utf-8") as f:
|
||||
first_line = f.readline().strip()
|
||||
# Cap the read so a malformed or maliciously crafted file (e.g.
|
||||
# a single huge line with no newline) can't blow up memory.
|
||||
first_line = f.readline(128).strip()
|
||||
value = "".join(c for c in first_line if 32 <= ord(c) < 127)
|
||||
if value:
|
||||
_cached_value = value
|
||||
return _cached_value
|
||||
return value
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
except Exception as e:
|
||||
logger.error("Failed to read %s: %s", env_file, e)
|
||||
|
||||
_cached_value = _DEFAULT_DEPLOY_ENV
|
||||
return _cached_value
|
||||
return _DEFAULT_DEPLOY_ENV
|
||||
|
||||
83
tests-unit/deploy_environment_test.py
Normal file
83
tests-unit/deploy_environment_test.py
Normal file
@ -0,0 +1,83 @@
|
||||
"""Tests for comfy.deploy_environment."""
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from comfy.deploy_environment import get_deploy_environment
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_cache_and_base_path(tmp_path, monkeypatch):
|
||||
"""Reset the functools cache and point folder_paths.base_path at a tmp dir for each test."""
|
||||
get_deploy_environment.cache_clear()
|
||||
import folder_paths
|
||||
monkeypatch.setattr(folder_paths, "base_path", str(tmp_path))
|
||||
yield
|
||||
get_deploy_environment.cache_clear()
|
||||
|
||||
|
||||
def _write_env_file(tmp_path, content: str) -> str:
|
||||
path = os.path.join(str(tmp_path), ".comfy_environment")
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
return path
|
||||
|
||||
|
||||
class TestGetDeployEnvironment:
|
||||
def test_returns_local_git_when_file_missing(self):
|
||||
assert get_deploy_environment() == "local-git"
|
||||
|
||||
def test_reads_value_from_file(self, tmp_path):
|
||||
_write_env_file(tmp_path, "local-desktop2-standalone\n")
|
||||
assert get_deploy_environment() == "local-desktop2-standalone"
|
||||
|
||||
def test_strips_trailing_whitespace_and_newline(self, tmp_path):
|
||||
_write_env_file(tmp_path, " local-desktop2-standalone \n")
|
||||
assert get_deploy_environment() == "local-desktop2-standalone"
|
||||
|
||||
def test_only_first_line_is_used(self, tmp_path):
|
||||
_write_env_file(tmp_path, "first-line\nsecond-line\n")
|
||||
assert get_deploy_environment() == "first-line"
|
||||
|
||||
def test_empty_file_falls_back_to_default(self, tmp_path):
|
||||
_write_env_file(tmp_path, "")
|
||||
assert get_deploy_environment() == "local-git"
|
||||
|
||||
def test_empty_after_whitespace_strip_falls_back_to_default(self, tmp_path):
|
||||
_write_env_file(tmp_path, " \n")
|
||||
assert get_deploy_environment() == "local-git"
|
||||
|
||||
def test_strips_control_chars_within_first_line(self, tmp_path):
|
||||
# Embedded NUL/control chars in the value should be stripped
|
||||
# (header-injection / smuggling protection).
|
||||
_write_env_file(tmp_path, "abc\x00\x07xyz\n")
|
||||
assert get_deploy_environment() == "abcxyz"
|
||||
|
||||
def test_strips_non_ascii_characters(self, tmp_path):
|
||||
_write_env_file(tmp_path, "café-é\n")
|
||||
assert get_deploy_environment() == "caf-"
|
||||
|
||||
def test_caps_read_at_128_bytes(self, tmp_path):
|
||||
# A single huge line with no newline must not be fully read into memory.
|
||||
huge = "x" * 10_000
|
||||
_write_env_file(tmp_path, huge)
|
||||
result = get_deploy_environment()
|
||||
assert result == "x" * 128
|
||||
|
||||
def test_result_is_cached_across_calls(self, tmp_path):
|
||||
path = _write_env_file(tmp_path, "first_value\n")
|
||||
assert get_deploy_environment() == "first_value"
|
||||
# Overwrite the file — cached value should still be returned.
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
f.write("second_value\n")
|
||||
assert get_deploy_environment() == "first_value"
|
||||
|
||||
def test_unreadable_file_falls_back_to_default(self, tmp_path, monkeypatch):
|
||||
_write_env_file(tmp_path, "should_not_be_used\n")
|
||||
|
||||
def _boom(*args, **kwargs):
|
||||
raise OSError("simulated read failure")
|
||||
|
||||
monkeypatch.setattr("builtins.open", _boom)
|
||||
assert get_deploy_environment() == "local-git"
|
||||
Loading…
Reference in New Issue
Block a user