test: expand HPC policy engine unit coverage

tests/_srcload.py

@@ -0,0 +1,44 @@
+"""
+Test-only helper: import a module directly from its .py source file on disk,
+bypassing normal package import resolution.
+
+Why this exists: setup.py (see EXTENSIONS there) Cython-compiles eight
+modules (app/core/gpu.py, policies.py, quota.py, scheduler.py,
+app/services/quota_service.py, scheduler_service.py, and
+app/api/routes_policy.py, routes_quota.py) into pre-built .so extensions
+that are checked into git alongside their .py source, e.g.
+app/core/gpu.cpython-313-aarch64-linux-gnu.so next to app/core/gpu.py.
+
+CPython's default import machinery prefers a matching extension module over
+a same-named .py file. On a host whose interpreter ABI/arch happens to match
+the checked-in .so tag (cpython-313-aarch64-linux-gnu), a plain
+`import app.core.gpu` silently resolves to the compiled .so, not the .py.
+coverage.py cannot trace execution inside a compiled extension, so on such a
+host those modules' lines never appear in a coverage report at all -- even
+though their logic *is* being exercised (indirectly, through the compiled
+binary) by tests that import them normally. On hosts where the .so tag does
+not match (e.g. this repo's CI, which runs Python 3.11), the .so is skipped
+automatically and the .py import already gets measured -- no workaround
+needed there.
+
+Loading the .py file directly by path (the same technique
+tests/test_setup_module.py already uses for setup.py) sidesteps the .so/.py
+shadowing so the real .py source is what gets executed and measured,
+regardless of host architecture. It does not change any production
+behavior -- Cython compiles these files essentially as-is, so the .py
+source and the compiled extension implement the same logic.
+"""
+import importlib.util
+import os
+
+_REPO_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
+
+
+def load(relative_path: str):
+    """Load and return the .py module at `relative_path` (repo-root-relative)."""
+    full_path = os.path.join(_REPO_ROOT, relative_path)
+    module_name = "srcload_" + relative_path.replace("/", "_").replace(".", "_")
+    spec = importlib.util.spec_from_file_location(module_name, full_path)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod

tests/test_config_module.py

@@ -0,0 +1,113 @@
+"""
+Unit tests for app/core/config.py::Config.
+
+Config.* are plain class attributes evaluated once, at module-exec time,
+from os.getenv(...). To exercise different env-var combinations
+deterministically (without mutating the single Config object every other
+test file in this session imports via `app.core.config`), each test here
+loads a *fresh* copy of config.py by file path (tests/_srcload.py) inside an
+os.environ patch, so the env vars are only visible to that one exec and
+nothing else in the suite is affected.
+"""
+import os
+from unittest.mock import patch
+
+from tests._srcload import load
+
+
+def _load_config_with_env(env: dict):
+    # os.environ is read at class-body execution time inside config.py, so
+    # the patch must be active for the load() call itself.
+    clean_env = {k: v for k, v in os.environ.items() if not k.startswith((
+        "MYSQL_", "REDIS_URL", "DEFAULT_CPU_HOURS", "DEFAULT_GPU_HOURS", "MAX_CONCURRENT_JOBS",
+    ))}
+    with patch.dict(os.environ, {**clean_env, **env}, clear=True):
+        return load("app/core/config.py")
+
+
+# ---------------------------------------------------------------------------
+# Safe defaults when nothing is configured
+# ---------------------------------------------------------------------------
+
+def test_defaults_used_when_no_env_vars_set():
+    cfg = _load_config_with_env({})
+    assert cfg.Config.MYSQL_HOST == "mysql"
+    assert cfg.Config.MYSQL_PORT == 3306
+    assert cfg.Config.MYSQL_DB == "omnibioai_hpc"
+    assert cfg.Config.MYSQL_USER == "root"
+    assert cfg.Config.MYSQL_PASSWORD == "root"
+    assert cfg.Config.REDIS_URL == "redis://redis:6379"
+    assert cfg.Config.DEFAULT_CPU_HOURS == 120
+    assert cfg.Config.DEFAULT_GPU_HOURS == 24
+    assert cfg.Config.MAX_CONCURRENT_JOBS == 5
+    assert cfg.Config.APP_NAME == "OmniBioAI HPC Policy Engine"
+
+
+# ---------------------------------------------------------------------------
+# Valid overrides
+# ---------------------------------------------------------------------------
+
+def test_env_vars_override_defaults():
+    cfg = _load_config_with_env({
+        "MYSQL_HOST": "db.internal",
+        "MYSQL_PORT": "5432",
+        "MYSQL_DB": "custom_db",
+        "MYSQL_USER": "svc",
+        "MYSQL_PASSWORD": "hunter2",
+        "REDIS_URL": "redis://cache:6380",
+        "DEFAULT_CPU_HOURS": "500",
+        "DEFAULT_GPU_HOURS": "50",
+        "MAX_CONCURRENT_JOBS": "20",
+    })
+    assert cfg.Config.MYSQL_HOST == "db.internal"
+    assert cfg.Config.MYSQL_PORT == 5432
+    assert cfg.Config.MYSQL_DB == "custom_db"
+    assert cfg.Config.MYSQL_USER == "svc"
+    assert cfg.Config.MYSQL_PASSWORD == "hunter2"
+    assert cfg.Config.REDIS_URL == "redis://cache:6380"
+    assert cfg.Config.DEFAULT_CPU_HOURS == 500
+    assert cfg.Config.DEFAULT_GPU_HOURS == 50
+    assert cfg.Config.MAX_CONCURRENT_JOBS == 20
+
+
+def test_zero_quota_defaults_are_respected_verbatim():
+    """A deliberately-zeroed quota env var is honored, not silently
+    replaced by a nonzero default."""
+    cfg = _load_config_with_env({"DEFAULT_CPU_HOURS": "0", "DEFAULT_GPU_HOURS": "0"})
+    assert cfg.Config.DEFAULT_CPU_HOURS == 0
+    assert cfg.Config.DEFAULT_GPU_HOURS == 0
+
+
+# ---------------------------------------------------------------------------
+# Invalid configuration -- no graceful handling exists (audit gap)
+# ---------------------------------------------------------------------------
+
+def test_non_numeric_mysql_port_raises_at_import_time():
+    """Characterizes current behavior: MYSQL_PORT is parsed with a bare
+    int(...) call and nothing catches a malformed value -- the module fails
+    to import at all (ValueError) rather than falling back to the default
+    or raising a clear configuration error. Documented as a "bugs
+    discovered but not fixed" item in the PR description; not fixed here
+    per the test-only scope of this change."""
+    import pytest
+    with pytest.raises(ValueError):
+        _load_config_with_env({"MYSQL_PORT": "not-a-port"})
+
+
+def test_non_numeric_default_cpu_hours_raises_at_import_time():
+    import pytest
+    with pytest.raises(ValueError):
+        _load_config_with_env({"DEFAULT_CPU_HOURS": "unlimited"})
+
+
+def test_non_numeric_max_concurrent_jobs_raises_at_import_time():
+    import pytest
+    with pytest.raises(ValueError):
+        _load_config_with_env({"MAX_CONCURRENT_JOBS": "many"})
+
+
+def test_empty_string_mysql_host_is_accepted_verbatim():
+    """Characterizes current behavior: string-typed settings have no
+    non-empty validation, so an explicitly-empty value is accepted as-is."""
+    cfg = _load_config_with_env({"MYSQL_HOST": ""})
+    assert cfg.Config.MYSQL_HOST == ""

tests/test_core_gpu_source.py

@@ -0,0 +1,64 @@
+"""
+Direct-source unit tests for app/core/gpu.py::validate_gpu_access.
+
+Loaded via tests/_srcload.py so its lines/branches are measured even on a
+host where the checked-in .so shadows the .py import (see _srcload.py for
+why). Behavior is already exercised indirectly via test_quota_service.py
+and test_routes_policy.py; this file targets the function directly with the
+full input space, including edge cases those higher-level tests don't hit.
+"""
+from tests._srcload import load
+
+gpu = load("app/core/gpu.py")
+
+
+def test_zero_gpus_needs_no_role():
+    ok, reason = gpu.validate_gpu_access([], 0)
+    assert ok is True
+    assert reason == "no gpu needed"
+
+
+def test_negative_gpus_short_circuits_to_no_gpu_needed():
+    """Characterizes current behavior: `gpus <= 0` is a single check, so a
+    negative gpu count is (silently) treated the same as zero/none."""
+    ok, reason = gpu.validate_gpu_access([], -3)
+    assert ok is True
+    assert reason == "no gpu needed"
+
+
+def test_positive_gpus_without_any_roles_denied():
+    ok, reason = gpu.validate_gpu_access([], 1)
+    assert ok is False
+    assert reason == "gpu access denied"
+
+
+def test_positive_gpus_with_unrelated_roles_denied():
+    ok, _reason = gpu.validate_gpu_access(["researcher", "viewer"], 1)
+    assert ok is False
+
+
+def test_positive_gpus_with_gpu_user_role_allowed():
+    ok, reason = gpu.validate_gpu_access(["researcher", "gpu_user"], 1)
+    assert ok is True
+    assert reason == "gpu allowed"
+
+
+def test_role_match_is_exact_not_substring():
+    """'gpu_user_temp' must not satisfy the 'gpu_user' membership check."""
+    ok, _reason = gpu.validate_gpu_access(["gpu_user_temp"], 1)
+    assert ok is False
+
+
+def test_role_match_is_case_sensitive():
+    ok, _reason = gpu.validate_gpu_access(["GPU_USER"], 1)
+    assert ok is False
+
+
+def test_large_gpu_request_with_role_allowed():
+    ok, _reason = gpu.validate_gpu_access(["gpu_user"], 64)
+    assert ok is True
+
+
+def test_empty_roles_list_with_zero_gpus_allowed():
+    ok, _reason = gpu.validate_gpu_access([], 0)
+    assert ok is True

tests/test_core_policies_source.py

@@ -0,0 +1,58 @@
+"""
+Direct-source unit tests for app/core/policies.py::validate_partition_access.
+
+Loaded via tests/_srcload.py (see that module's docstring) so its lines are
+measured even where the checked-in .so shadows the .py import.
+"""
+from tests._srcload import load
+
+policies = load("app/core/policies.py")
+
+
+def test_dgx_partition_denied_without_dgx_access_role():
+    ok, reason = policies.validate_partition_access([], "dgx-a100")
+    assert ok is False
+    assert reason == "dgx partition denied"
+
+
+def test_dgx_partition_denied_with_unrelated_roles():
+    ok, _reason = policies.validate_partition_access(["gpu_user", "researcher"], "dgx-a100")
+    assert ok is False
+
+
+def test_dgx_partition_allowed_with_dgx_access_role():
+    ok, reason = policies.validate_partition_access(["dgx_access"], "dgx-a100")
+    assert ok is True
+    assert reason == "partition allowed"
+
+
+def test_cpu_partition_allowed_with_no_roles():
+    ok, _reason = policies.validate_partition_access([], "cpu")
+    assert ok is True
+
+
+def test_gpu_partition_allowed_with_no_roles():
+    """Only the literal 'dgx-a100' partition is gated -- any other partition
+    name (including 'gpu') is allowed regardless of roles."""
+    ok, _reason = policies.validate_partition_access([], "gpu")
+    assert ok is True
+
+
+def test_unknown_partition_name_allowed_by_default():
+    """Characterizes current behavior: partition names aren't validated
+    against an allow-list, so an unrecognized/typo'd partition passes
+    through as allowed rather than being rejected."""
+    ok, _reason = policies.validate_partition_access([], "totally-made-up-partition")
+    assert ok is True
+
+
+def test_empty_partition_string_allowed():
+    ok, _reason = policies.validate_partition_access([], "")
+    assert ok is True
+
+
+def test_partition_check_is_case_sensitive():
+    """'DGX-A100' does not match the literal 'dgx-a100' gate, so it's
+    treated as an ungated partition and allowed."""
+    ok, _reason = policies.validate_partition_access([], "DGX-A100")
+    assert ok is True