chore: generated test and production code for user-story-for-python-dotnet-for-env-check.md

wubin28 · wubin28 · commit 68376673393f · 2026-03-07T16:32:13.000+08:00
diff --git a/dialogs/dialog-2026-03-07--16-30.md b/dialogs/dialog-2026-03-07--16-30.md
diff --git a/env-check/env_check.py b/env-check/env_check.py
@@ -0,0 +1,77 @@
+import os
+import sys
+from dotenv import load_dotenv
+
+
+VAR_SPECS = [
+    {"name": "DB_HOST",     "type": "str"},
+    {"name": "DB_PORT",     "type": "int",  "min": 1, "max": 65535},
+    {"name": "DB_NAME",     "type": "str"},
+    {"name": "DB_USER",     "type": "str"},
+    {"name": "DB_PASSWORD", "type": "str"},
+    {"name": "BATCH_SIZE",  "type": "int",  "min": 1, "range_desc": "BATCH_SIZE > 0"},
+    {"name": "MAX_RETRIES", "type": "int",  "min": 0, "range_desc": "MAX_RETRIES >= 0"},
+    {"name": "DRY_RUN",     "type": "bool"},
+]
+
+VALID_BOOLS = frozenset({"true", "false", "1", "0"})
+
+
+def _check_int(name: str, value: str, spec: dict) -> str | None:
+    try:
+        int_val = int(value)
+    except ValueError:
+        return f'[FAIL] {name}: 无法转换为整数（当前值："{value}"）'
+
+    lo, hi = spec.get("min"), spec.get("max")
+    if hi is not None:
+        if not (lo <= int_val <= hi):
+            return f"[FAIL] {name}: 值超出范围（需满足 {lo} <= {name} <= {hi}，当前值：{int_val}）"
+    elif lo is not None and int_val < lo:
+        range_desc = spec.get("range_desc", f"{name} >= {lo}")
+        return f"[FAIL] {name}: 值超出范围（需满足 {range_desc}）"
+    return None
+
+
+def collect_errors(env_vars: dict) -> list[str]:
+    errors = []
+    for spec in VAR_SPECS:
+        name = spec["name"]
+        value = env_vars.get(name)
+
+        if value is None:
+            errors.append(f"[FAIL] {name}: 未设置")
+            continue
+
+        if spec["type"] == "str":
+            if value == "":
+                errors.append(f"[FAIL] {name}: 值为空字符串")
+        elif spec["type"] == "int":
+            err = _check_int(name, value, spec)
+            if err:
+                errors.append(err)
+        else:  # bool
+            if value.lower() not in VALID_BOOLS:
+                errors.append(f'[FAIL] {name}: 非法布尔值（当前值："{value}"，合法值为 true/false/1/0）')
+
+    return errors
+
+
+def validate_env(dotenv_path=None) -> list[str]:
+    load_dotenv(dotenv_path=dotenv_path, override=True)
+    return collect_errors(os.environ)
+
+
+def main():
+    errors = validate_env()
+    if errors:
+        for msg in errors:
+            print(msg)
+        sys.exit(1)
+    else:
+        print("[OK] 所有环境变量校验通过")
+        sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/env-check/pytest.ini b/env-check/pytest.ini
@@ -0,0 +1,2 @@
+[pytest]
+testpaths = test
diff --git a/env-check/test/__init__.py b/env-check/test/__init__.py
diff --git a/env-check/test/conftest.py b/env-check/test/conftest.py
@@ -0,0 +1,22 @@
+import os
+import sys
+from pathlib import Path
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+REQUIRED_VAR_NAMES = [
+    "DB_HOST", "DB_PORT", "DB_NAME", "DB_USER", "DB_PASSWORD",
+    "BATCH_SIZE", "MAX_RETRIES", "DRY_RUN",
+]
+
+@pytest.fixture
+def clean_env():
+    """Remove required env vars before test; restore original state after."""
+    saved = {name: os.environ.pop(name, None) for name in REQUIRED_VAR_NAMES}
+    yield
+    for name, original_value in saved.items():
+        if original_value is None:
+            os.environ.pop(name, None)
+        else:
+            os.environ[name] = original_value
diff --git a/env-check/test/test_env_check.py b/env-check/test/test_env_check.py
@@ -0,0 +1,129 @@
+import pytest
+from pathlib import Path
+from env_check import validate_env
+
+VALID_ENV_CONTENT = (
+    "DB_HOST=localhost\n"
+    "DB_PORT=5432\n"
+    "DB_NAME=analytics\n"
+    "DB_USER=admin\n"
+    "DB_PASSWORD=s3cret\n"
+    "BATCH_SIZE=500\n"
+    "MAX_RETRIES=3\n"
+    "DRY_RUN=false\n"
+)
+
+
+# AC-H1: All vars valid → no errors, exit 0
+def test_h1_all_valid(clean_env, tmp_path):
+    (tmp_path / ".env").write_text(VALID_ENV_CONTENT, encoding="utf-8")
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert errors == []
+
+
+# AC-H2: DRY_RUN accepts true/True/TRUE/1/false/False/0
+@pytest.mark.parametrize("value", ["true", "True", "TRUE", "1", "false", "False", "0"])
+def test_h2_dry_run_valid_values(clean_env, tmp_path, value):
+    content = VALID_ENV_CONTENT.replace("DRY_RUN=false", f"DRY_RUN={value}")
+    (tmp_path / ".env").write_text(content, encoding="utf-8")
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert errors == []
+
+
+# AC-H3: DB_PORT boundary values 1 and 65535
+@pytest.mark.parametrize("port", [1, 65535])
+def test_h3_db_port_boundary(clean_env, tmp_path, port):
+    content = VALID_ENV_CONTENT.replace("DB_PORT=5432", f"DB_PORT={port}")
+    (tmp_path / ".env").write_text(content, encoding="utf-8")
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert errors == []
+
+
+# AC-H4: MAX_RETRIES=0 → OK
+def test_h4_max_retries_zero(clean_env, tmp_path):
+    content = VALID_ENV_CONTENT.replace("MAX_RETRIES=3", "MAX_RETRIES=0")
+    (tmp_path / ".env").write_text(content, encoding="utf-8")
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert errors == []
+
+
+# AC-S1: .env file doesn't exist → 8 FAIL lines, exit 1
+def test_s1_no_env_file(clean_env, tmp_path):
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert len(errors) == 8
+    assert all(e.startswith("[FAIL]") for e in errors)
+
+
+# AC-S2: DB_HOST missing
+def test_s2_db_host_missing(clean_env, tmp_path):
+    content = "\n".join(
+        line for line in VALID_ENV_CONTENT.splitlines()
+        if not line.startswith("DB_HOST=")
+    ) + "\n"
+    (tmp_path / ".env").write_text(content, encoding="utf-8")
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert "[FAIL] DB_HOST: 未设置" in errors
+
+
+# AC-S3: DB_PASSWORD empty string
+def test_s3_db_password_empty(clean_env, tmp_path):
+    content = VALID_ENV_CONTENT.replace("DB_PASSWORD=s3cret", "DB_PASSWORD=")
+    (tmp_path / ".env").write_text(content, encoding="utf-8")
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert "[FAIL] DB_PASSWORD: 值为空字符串" in errors
+
+
+# AC-S4: DB_PORT non-integer
+def test_s4_db_port_not_int(clean_env, tmp_path):
+    content = VALID_ENV_CONTENT.replace("DB_PORT=5432", "DB_PORT=abc")
+    (tmp_path / ".env").write_text(content, encoding="utf-8")
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert '[FAIL] DB_PORT: 无法转换为整数（当前值："abc"）' in errors
+
+
+# AC-S5: DB_PORT out of range (0 and 65536)
+@pytest.mark.parametrize("port,expected_val", [(0, 0), (65536, 65536)])
+def test_s5_db_port_out_of_range(clean_env, tmp_path, port, expected_val):
+    content = VALID_ENV_CONTENT.replace("DB_PORT=5432", f"DB_PORT={port}")
+    (tmp_path / ".env").write_text(content, encoding="utf-8")
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert f"[FAIL] DB_PORT: 值超出范围（需满足 1 <= DB_PORT <= 65535，当前值：{expected_val}）" in errors
+
+
+# AC-S6: BATCH_SIZE=0 and BATCH_SIZE=-1
+@pytest.mark.parametrize("size", [0, -1])
+def test_s6_batch_size_not_positive(clean_env, tmp_path, size):
+    content = VALID_ENV_CONTENT.replace("BATCH_SIZE=500", f"BATCH_SIZE={size}")
+    (tmp_path / ".env").write_text(content, encoding="utf-8")
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert "[FAIL] BATCH_SIZE: 值超出范围（需满足 BATCH_SIZE > 0）" in errors
+
+
+# AC-S7: DRY_RUN=yes → invalid bool
+def test_s7_dry_run_invalid(clean_env, tmp_path):
+    content = VALID_ENV_CONTENT.replace("DRY_RUN=false", "DRY_RUN=yes")
+    (tmp_path / ".env").write_text(content, encoding="utf-8")
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert '[FAIL] DRY_RUN: 非法布尔值（当前值："yes"，合法值为 true/false/1/0）' in errors
+
+
+# AC-S8: Multiple errors in order
+def test_s8_multiple_errors_in_order(clean_env, tmp_path):
+    content = (
+        "DB_HOST=localhost\n"
+        "DB_PORT=abc\n"
+        "DB_NAME=analytics\n"
+        "DB_USER=admin\n"
+        "DB_PASSWORD=\n"
+        "BATCH_SIZE=-1\n"
+        "MAX_RETRIES=3\n"
+        "DRY_RUN=yes\n"
+    )
+    (tmp_path / ".env").write_text(content, encoding="utf-8")
+    errors = validate_env(dotenv_path=tmp_path / ".env")
+    assert len(errors) == 4
+    assert '[FAIL] DB_PORT: 无法转换为整数（当前值："abc"）' == errors[0]
+    assert "[FAIL] DB_PASSWORD: 值为空字符串" == errors[1]
+    assert "[FAIL] BATCH_SIZE: 值超出范围（需满足 BATCH_SIZE > 0）" == errors[2]
+    assert '[FAIL] DRY_RUN: 非法布尔值（当前值："yes"，合法值为 true/false/1/0）' == errors[3]
+
