|
# retoor <retoor@molodetz.nl>
|
|
|
|
import pytest
|
|
|
|
from devplacepy.database import get_table, init_db, set_setting
|
|
from devplacepy.services.containers.workspace import quota
|
|
from devplacepy.utils import generate_uid
|
|
|
|
OWNER = "quota-owner"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _quota_db():
|
|
init_db()
|
|
yield
|
|
get_table(quota.RULES_TABLE).delete()
|
|
for key in quota.SETTING_KEYS.values():
|
|
set_setting(key, str(quota.DEFAULTS[_key_for(key)]))
|
|
|
|
|
|
def _key_for(setting: str) -> str:
|
|
for key, value in quota.SETTING_KEYS.items():
|
|
if value == setting:
|
|
return key
|
|
raise KeyError(setting)
|
|
|
|
|
|
def _rule(**values) -> dict:
|
|
row = {
|
|
"uid": generate_uid(),
|
|
"owner_kind": "user",
|
|
"owner_id": OWNER,
|
|
"label": "test",
|
|
"created_at": "",
|
|
"updated_at": "",
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
**{column: 0 for column in quota.RULE_COLUMNS},
|
|
}
|
|
row.update(values)
|
|
get_table(quota.RULES_TABLE).insert(row)
|
|
return row
|
|
|
|
|
|
def test_format_cpu_renders_docker_values():
|
|
assert quota.format_cpu(2000) == "2"
|
|
assert quota.format_cpu(1500) == "1.5"
|
|
assert quota.format_cpu(250) == "0.25"
|
|
assert quota.format_cpu(0) == ""
|
|
assert quota.format_cpu(-1) == ""
|
|
|
|
|
|
def test_format_memory_renders_docker_values():
|
|
assert quota.format_memory(2048) == "2048m"
|
|
assert quota.format_memory(0) == ""
|
|
|
|
|
|
def test_limits_expose_the_docker_strings():
|
|
limits = quota.resolve()
|
|
assert limits.cpu_limit() == quota.format_cpu(limits.cpu_millicores)
|
|
assert limits.mem_limit() == quota.format_memory(limits.memory_mb)
|
|
|
|
|
|
def test_defaults_resolve_without_a_rule():
|
|
limits = quota.resolve(OWNER)
|
|
assert limits.cpu_millicores == quota.DEFAULTS["cpu_millicores"]
|
|
assert limits.memory_mb == quota.DEFAULTS["memory_mb"]
|
|
|
|
|
|
def test_a_user_rule_resolves():
|
|
_rule(cpu_millicores=4000, memory_mb=8192, disk_quota_mb=512)
|
|
limits = quota.resolve(OWNER)
|
|
assert limits.cpu_millicores == 4000
|
|
assert limits.memory_mb == 8192
|
|
assert limits.disk_quota_mb == 512
|
|
|
|
|
|
def test_a_user_rule_does_not_leak_to_another_user():
|
|
_rule(cpu_millicores=4000)
|
|
assert quota.resolve("someone-else").cpu_millicores == (
|
|
quota.DEFAULTS["cpu_millicores"]
|
|
)
|
|
|
|
|
|
def test_a_soft_deleted_rule_is_ignored():
|
|
row = _rule(cpu_millicores=4000)
|
|
get_table(quota.RULES_TABLE).update(
|
|
{"uid": row["uid"], "deleted_at": "2026-01-01T00:00:00+00:00"}, ["uid"]
|
|
)
|
|
assert quota.resolve(OWNER).cpu_millicores == quota.DEFAULTS["cpu_millicores"]
|
|
|
|
|
|
def test_an_instance_override_beats_the_rule():
|
|
_rule(cpu_millicores=4000, memory_mb=8192)
|
|
limits = quota.resolve(
|
|
OWNER, {"workspace_cpu_millicores": 1000, "workspace_memory_mb": 512}
|
|
)
|
|
assert limits.cpu_millicores == 1000
|
|
assert limits.memory_mb == 512
|
|
|
|
|
|
def test_only_declared_instance_overrides_are_read():
|
|
limits = quota.resolve(OWNER, {"workspace_max_tunnels": 99})
|
|
assert limits.max_tunnels == quota.DEFAULTS["max_tunnels"]
|
|
assert "max_tunnels" not in quota.INSTANCE_OVERRIDE_COLUMNS
|
|
|
|
|
|
def test_the_idle_warning_stays_below_the_idle_stop():
|
|
set_setting("workspace_idle_stop_minutes", "10")
|
|
set_setting("workspace_idle_warn_minutes", "30")
|
|
limits = quota.resolve()
|
|
assert limits.idle_warn_minutes < limits.idle_stop_minutes
|