mteehan
78fcb01877
- install.sh: the traversal-chmod loop assigned _d but looped over the never-set $d; under set -u every fresh install aborted with "d: unbound variable" at that line. Loop over $_d. - acme collector: the self-heal normalize (sudo chmod g+rwX) now runs only when a no-sudo group-read-bit probe detects a lost bit — acme.sh re-hardens the tree 600 on every run, so the steady-state poll makes no sudo call. The group bit (not daemon readability) is what the two-user model keeps for the WebUI user. - lib.acme: new get_acme_home() accessor (ACME_HOME env, default data/acme), reused by _run_acme; _summarize_acme_output preserves a "Permission denied" line even when it is not among the final two, so the collector's actionable-error matcher keeps firing. - nginx template: emit location /static/ for any is_management path (not only '/'); the SPA references /static/... at the domain root regardless of the management backend path. - tests: probe, summarizer, and nginx-subpath cases in test_state.py, test_acme.py, test_nginx.py.
1090 lines
40 KiB
Python
1090 lines
40 KiB
Python
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from lib import nginx
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_default_config():
|
|
"""Reset the shared mutable DEFAULT_CONFIG before each test."""
|
|
original = nginx.DEFAULT_CONFIG.copy()
|
|
yield
|
|
# Reset the shared "domains" dict that leaks due to shallow copy in _json_load
|
|
nginx.DEFAULT_CONFIG = original
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_data_dir(tmp_path):
|
|
original_config = nginx.CONFIG_FILE
|
|
original_sites = nginx.SITES_DIR
|
|
original_htpasswd = nginx.HTPASSWD_FILE
|
|
original_ssl_snippet = nginx.SSL_SNIPPET
|
|
original_include = nginx.INCLUDE_FILE
|
|
original_config_dir = nginx.CONFIG_DIR
|
|
original_data_dir = nginx.DATA_DIR
|
|
|
|
nginx.CONFIG_DIR = tmp_path / "nginx"
|
|
nginx.DATA_DIR = tmp_path / "nginx"
|
|
nginx.SITES_DIR = tmp_path / "nginx" / "sites-enabled"
|
|
nginx.CONFIG_FILE = tmp_path / "nginx" / "config.json"
|
|
nginx.HTPASSWD_FILE = tmp_path / "nginx" / ".htpasswd"
|
|
nginx.SSL_SNIPPET = tmp_path / "ssl_snippet.conf"
|
|
nginx.INCLUDE_FILE = tmp_path / "include.conf"
|
|
|
|
nginx.CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
|
nginx.DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
nginx.SITES_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
yield tmp_path
|
|
|
|
nginx.CONFIG_FILE = original_config
|
|
nginx.SITES_DIR = original_sites
|
|
nginx.HTPASSWD_FILE = original_htpasswd
|
|
nginx.SSL_SNIPPET = original_ssl_snippet
|
|
nginx.INCLUDE_FILE = original_include
|
|
nginx.CONFIG_DIR = original_config_dir
|
|
nginx.DATA_DIR = original_data_dir
|
|
|
|
|
|
class TestGetConfig:
|
|
def test_returns_default_when_no_file(self, temp_data_dir):
|
|
cfg = nginx.get_config()
|
|
assert "domains" in cfg
|
|
assert "ssl" in cfg
|
|
assert cfg["domains"] == {}
|
|
|
|
def test_read_does_not_rewrite_unchanged_file(self, temp_data_dir):
|
|
"""get_config() must not re-save a file that needs no migration."""
|
|
nginx.save_config(
|
|
{
|
|
"backends": {"webui": {"_migrated": True, "paths": {}}},
|
|
"domains": {"app.example.com": {"backend": "webui"}},
|
|
"ssl": {"protocols": "TLSv1.3"},
|
|
}
|
|
)
|
|
mtime_before = nginx.CONFIG_FILE.stat().st_mtime_ns
|
|
cfg = nginx.get_config()
|
|
assert cfg["domains"] == {"app.example.com": {"backend": "webui"}}
|
|
# No churn: reading a current-format config leaves the file alone.
|
|
assert nginx.CONFIG_FILE.stat().st_mtime_ns == mtime_before
|
|
|
|
def test_read_migrates_in_memory_without_writing(self, temp_data_dir):
|
|
"""get_config() is pure: migration is applied in memory, file untouched."""
|
|
nginx.save_config({"domains": {"app.example.com": {"backend": "myapp"}}})
|
|
mtime_before = nginx.CONFIG_FILE.stat().st_mtime_ns
|
|
cfg = nginx.get_config()
|
|
# Migration added the builtin webui backend (in memory only).
|
|
assert cfg["backends"]["webui"]["_migrated"] is True
|
|
assert nginx.CONFIG_FILE.stat().st_mtime_ns == mtime_before
|
|
|
|
def test_migrate_config_file_persists_legacy(self, temp_data_dir):
|
|
"""migrate_config_file() rewrites the file when migration changes it."""
|
|
nginx.save_config({"domains": {"app.example.com": {"backend": "myapp"}}})
|
|
mtime_before = nginx.CONFIG_FILE.stat().st_mtime_ns
|
|
assert nginx.migrate_config_file() is True
|
|
assert nginx.CONFIG_FILE.stat().st_mtime_ns != mtime_before
|
|
# Idempotent: a second run is a no-op.
|
|
assert nginx.migrate_config_file() is False
|
|
|
|
def test_migrate_config_file_noop_when_missing(self, temp_data_dir):
|
|
assert not nginx.CONFIG_FILE.exists()
|
|
assert nginx.migrate_config_file() is False
|
|
assert not nginx.CONFIG_FILE.exists()
|
|
|
|
|
|
class TestSaveConfig:
|
|
def test_saves_and_reloads(self, temp_data_dir):
|
|
cfg = {
|
|
"backends": {
|
|
"myapp": {
|
|
"label": "My App",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {
|
|
"host": "localhost",
|
|
"port": 80,
|
|
"proto": "http",
|
|
}
|
|
}
|
|
},
|
|
}
|
|
},
|
|
"domains": {"example.com": {"backend": "myapp"}},
|
|
}
|
|
nginx.save_config(cfg)
|
|
loaded = nginx.get_config()
|
|
assert loaded["domains"]["example.com"]["backend"] == "myapp"
|
|
assert (
|
|
loaded["backends"]["myapp"]["paths"]["/"]["backend"]["host"] == "localhost"
|
|
)
|
|
|
|
|
|
class TestGetDomains:
|
|
def test_empty_domains(self, temp_data_dir):
|
|
result = nginx.get_domains()
|
|
assert result == []
|
|
|
|
def test_returns_domain_list(self, temp_data_dir):
|
|
cfg = {
|
|
"backends": {
|
|
"myapp": {
|
|
"label": "My App",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {
|
|
"host": "localhost",
|
|
"port": 8080,
|
|
"proto": "http",
|
|
}
|
|
},
|
|
"/api": {
|
|
"backend": {
|
|
"host": "localhost",
|
|
"port": 8081,
|
|
"proto": "http",
|
|
}
|
|
},
|
|
},
|
|
}
|
|
},
|
|
"domains": {
|
|
"example.com": {"backend": "myapp", "force_ssl": True},
|
|
},
|
|
}
|
|
nginx.save_config(cfg)
|
|
result = nginx.get_domains()
|
|
assert len(result) == 2
|
|
assert result[0]["domain"] == "example.com"
|
|
assert result[0]["backend_name"] == "myapp"
|
|
assert result[0]["path"] == "/"
|
|
|
|
|
|
class TestAddDomain:
|
|
def test_add_domain(self, temp_data_dir):
|
|
cfg = {
|
|
"backends": {
|
|
"myapp": {
|
|
"label": "My App",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {
|
|
"host": "10.0.0.5",
|
|
"port": 8080,
|
|
"proto": "http",
|
|
}
|
|
}
|
|
},
|
|
}
|
|
},
|
|
"domains": {},
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.save_config(cfg)
|
|
nginx.add_domain("example.com", "myapp")
|
|
loaded = nginx.get_config()
|
|
assert "example.com" in loaded["domains"]
|
|
assert loaded["domains"]["example.com"]["backend"] == "myapp"
|
|
assert loaded["domains"]["example.com"]["force_ssl"] is True
|
|
|
|
@patch("lib.nginx.get_config")
|
|
def test_duplicate_domain_raises(self, mock_get, temp_data_dir):
|
|
mock_get.return_value = {
|
|
"backends": {"myapp": {}},
|
|
"domains": {"example.com": {"backend": "myapp"}},
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.save_config(mock_get.return_value)
|
|
with pytest.raises(ValueError):
|
|
nginx.add_domain("example.com", "myapp")
|
|
|
|
|
|
class TestRemoveDomain:
|
|
@patch("lib.nginx.get_config")
|
|
def test_remove_existing_domain(self, mock_get, temp_data_dir):
|
|
mock_get.return_value = {
|
|
"domains": {
|
|
"example.com": {"backend": {"host": "x", "port": 80, "proto": "http"}}
|
|
},
|
|
"management": None,
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.remove_domain("example.com")
|
|
cfg = nginx.get_config()
|
|
assert "example.com" not in cfg["domains"]
|
|
|
|
def test_remove_nonexistent_domain(self, temp_data_dir):
|
|
nginx.remove_domain("nonexistent.com")
|
|
cfg = nginx.get_config()
|
|
assert "nonexistent.com" not in cfg["domains"]
|
|
|
|
|
|
class TestUpdateDomain:
|
|
@patch("lib.nginx.get_config")
|
|
def test_update_existing_domain(self, mock_get, temp_data_dir):
|
|
entry = {
|
|
"backend": {"host": "10.0.0.5", "port": 8080, "proto": "http"},
|
|
"force_ssl": True,
|
|
}
|
|
mock_get.return_value = {
|
|
"domains": {"example.com": entry},
|
|
"management": None,
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.update_domain("example.com", force_ssl=False)
|
|
cfg = nginx.get_config()
|
|
assert cfg["domains"]["example.com"]["force_ssl"] is False
|
|
|
|
def test_update_nonexistent_raises(self, temp_data_dir):
|
|
with pytest.raises(KeyError):
|
|
nginx.update_domain("nonexistent.com", force_ssl=False)
|
|
|
|
|
|
class TestWriteSite:
|
|
def test_write_creates_file(self, temp_data_dir):
|
|
nginx.write_site("example.com", "server { listen 443; }")
|
|
path = nginx.SITES_DIR / "example.com.conf"
|
|
assert path.exists()
|
|
content = Path(path).read_text()
|
|
assert "server { listen 443; }" in content
|
|
|
|
|
|
class TestGenerateServerConf:
|
|
def test_simple_root_path(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "10.0.0.1", "port": 80, "proto": "http"},
|
|
"headers": {"X-Custom": "value"},
|
|
}
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "location /" in out
|
|
assert "proxy_pass http://10.0.0.1:80;" in out
|
|
assert "proxy_set_header X-Custom value;" in out
|
|
assert "add_header X-Content-Type-Options" in out
|
|
|
|
def test_multiple_paths(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "app.example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "10.0.0.1", "port": 80, "proto": "http"},
|
|
"headers": {},
|
|
},
|
|
"/api": {
|
|
"backend": {"host": "10.0.0.2", "port": 8080, "proto": "http"},
|
|
},
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "proxy_pass http://10.0.0.1:80;" in out
|
|
assert "proxy_pass http://10.0.0.2:8080;" in out
|
|
assert "location /api" in out
|
|
|
|
def test_management_path(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "mgmt.example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "127.0.0.1", "port": 9090, "proto": "http"},
|
|
"is_management": True,
|
|
}
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
"auth": {"user": "admin", "htpasswd": "/path/.htpasswd"},
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "proxy_pass http://127.0.0.1:9090;" in out
|
|
# Server-level security headers come from Flask, not nginx
|
|
assert "Strict-Transport-Security" not in out
|
|
assert "Referrer-Policy" not in out
|
|
assert "wall_mgmt_access.log" in out
|
|
|
|
def test_management_static_location(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "mgmt.example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "127.0.0.1", "port": 9090, "proto": "http"},
|
|
"is_management": True,
|
|
},
|
|
"/ws": {
|
|
"backend": {"host": "127.0.0.1", "port": 9091, "proto": "http"},
|
|
"is_websocket": True,
|
|
},
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
static_root = str(nginx.PROJECT_DIR / "webui" / "static")
|
|
assert "location /static/ {" in out
|
|
assert f"alias {static_root}/;" in out
|
|
assert 'add_header Cache-Control "no-cache" always;' in out
|
|
assert "add_header X-Content-Type-Options nosniff always;" in out
|
|
assert (
|
|
"add_header Content-Security-Policy \"default-src 'none'\" always;" in out
|
|
)
|
|
|
|
def test_static_location_only_for_management_root(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "app.example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "10.0.0.1", "port": 80, "proto": "http"},
|
|
}
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "location /static/" not in out
|
|
|
|
def test_management_static_location_on_subpath(self, temp_data_dir):
|
|
# The SPA references /static/... at the domain root regardless of the
|
|
# management backend path, so the block is emitted for any
|
|
# is_management path, not only '/'.
|
|
cfg = {
|
|
"domain": "mgmt.example.com",
|
|
"paths": {
|
|
"/app": {
|
|
"backend": {"host": "127.0.0.1", "port": 9090, "proto": "http"},
|
|
"is_management": True,
|
|
},
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "location /static/ {" in out
|
|
static_root = str(nginx.PROJECT_DIR / "webui" / "static")
|
|
assert f"alias {static_root}/;" in out
|
|
|
|
def test_websocket_path(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "mgmt.example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "127.0.0.1", "port": 9090, "proto": "http"},
|
|
},
|
|
"/ws": {
|
|
"backend": {"host": "127.0.0.1", "port": 9091, "proto": "http"},
|
|
"is_websocket": True,
|
|
},
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "proxy_pass http://127.0.0.1:9091;" in out
|
|
assert "proxy_set_header Upgrade" in out
|
|
assert "proxy_read_timeout 86400s;" in out
|
|
|
|
def test_auth_inheritance(self, temp_data_dir):
|
|
cfg = {
|
|
"domain": "app.example.com",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "10.0.0.1", "port": 80, "proto": "http"},
|
|
"headers": {},
|
|
},
|
|
"/api": {
|
|
"backend": {"host": "10.0.0.2", "port": 8080, "proto": "http"},
|
|
"auth": None,
|
|
},
|
|
"/admin": {
|
|
"backend": {"host": "10.0.0.3", "port": 9000, "proto": "http"},
|
|
"auth": {"user": "admin", "htpasswd": "/other/.htpasswd"},
|
|
},
|
|
},
|
|
"force_ssl": True,
|
|
"cert": "acme",
|
|
"auth": {"user": "admin", "htpasswd": "/path/.htpasswd"},
|
|
}
|
|
out = nginx.generate_server_conf(cfg)
|
|
assert "auth_basic_user_file /path/.htpasswd;" in out
|
|
lines = out.split("\n")
|
|
api_idx = next(i for i, line in enumerate(lines) if "location /api" in line)
|
|
admin_idx = next(i for i, line in enumerate(lines) if "location /admin" in line)
|
|
# /api should have auth_basic off
|
|
assert "auth_basic off;" in "\n".join(lines[api_idx : api_idx + 5])
|
|
# /admin should have path-level auth override
|
|
assert "auth_basic_user_file /other/.htpasswd;" in "\n".join(
|
|
lines[admin_idx : admin_idx + 5]
|
|
)
|
|
|
|
|
|
class TestWriteAllSites:
|
|
@patch("lib.nginx.get_config")
|
|
def test_writes_all_domains(self, mock_get, temp_data_dir):
|
|
mock_get.return_value = {
|
|
"domains": {
|
|
"a.com": {
|
|
"paths": {
|
|
"/": {
|
|
"backend": {
|
|
"host": "10.0.0.1",
|
|
"port": 80,
|
|
"proto": "http",
|
|
},
|
|
"headers": {},
|
|
}
|
|
},
|
|
"force_ssl": True,
|
|
},
|
|
"b.com": {
|
|
"paths": {
|
|
"/": {
|
|
"backend": {
|
|
"host": "10.0.0.2",
|
|
"port": 80,
|
|
"proto": "http",
|
|
},
|
|
"headers": {},
|
|
}
|
|
},
|
|
"force_ssl": True,
|
|
},
|
|
},
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.write_all_sites()
|
|
assert (nginx.SITES_DIR / "a.com.conf").exists()
|
|
assert (nginx.SITES_DIR / "b.com.conf").exists()
|
|
|
|
@patch("lib.nginx.get_config")
|
|
def test_removes_old_sites(self, mock_get, temp_data_dir):
|
|
# Pre-create an old site
|
|
nginx.write_site("old.com", "server {}")
|
|
assert (nginx.SITES_DIR / "old.com.conf").exists()
|
|
|
|
mock_get.return_value = {
|
|
"domains": {},
|
|
"management": None,
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.write_all_sites()
|
|
assert not (nginx.SITES_DIR / "old.com.conf").exists()
|
|
|
|
|
|
class TestTestConfig:
|
|
@patch("lib.nginx.subprocess.run")
|
|
def test_passes(self, mock_run, temp_data_dir):
|
|
mock_run.return_value = MagicMock(
|
|
returncode=0, stdout="", stderr="test passed\n"
|
|
)
|
|
ok, _msg = nginx.test_config()
|
|
assert ok is True
|
|
|
|
@patch("lib.nginx.subprocess.run")
|
|
def test_fails(self, mock_run, temp_data_dir):
|
|
mock_run.return_value = MagicMock(
|
|
returncode=1, stdout="", stderr="nginx: configuration test failed\n"
|
|
)
|
|
ok, _msg = nginx.test_config()
|
|
assert ok is False
|
|
|
|
|
|
class TestWriteHtpasswd:
|
|
@patch("lib.nginx._hash_password")
|
|
def test_creates_file(self, mock_hash, temp_data_dir):
|
|
mock_hash.return_value = "$apr1$hash"
|
|
nginx.write_htpasswd("admin", "secret")
|
|
assert nginx.HTPASSWD_FILE.exists()
|
|
content = nginx.HTPASSWD_FILE.read_text()
|
|
assert "admin:" in content
|
|
|
|
@patch("lib.nginx._hash_password")
|
|
def test_replaces_existing_user(self, mock_hash, temp_data_dir):
|
|
mock_hash.return_value = "$apr1$hash1"
|
|
nginx.write_htpasswd("admin", "old")
|
|
mock_hash.return_value = "$apr1$hash2"
|
|
nginx.write_htpasswd("admin", "new")
|
|
lines = [
|
|
line
|
|
for line in nginx.HTPASSWD_FILE.read_text().strip().splitlines()
|
|
if line
|
|
]
|
|
assert len([line for line in lines if line.startswith("admin:")]) == 1
|
|
|
|
|
|
class TestHashPasswordFallback:
|
|
@patch("lib.nginx._hash_password")
|
|
def test_hash_returns_string(self, mock_hash, temp_data_dir):
|
|
mock_hash.return_value = "$apr1$hash"
|
|
result = nginx._hash_password("test")
|
|
assert isinstance(result, str)
|
|
assert len(result) > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Backend CRUD and resolution tests (daemon handler)
|
|
# NOTE: The daemon handler modules cannot be called as-is because they
|
|
# require sudo and live service paths. Instead we test through `lib.nginx`
|
|
# public API where possible, and unit-test the handler-internal functions
|
|
# by importing them directly.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestResolvePaths:
|
|
def test_resolves_from_backend(self, temp_data_dir):
|
|
"""Paths resolve from backends[domain_cfg['backend']].paths."""
|
|
backends = {
|
|
"myapp": {
|
|
"label": "My App",
|
|
"paths": {
|
|
"/": {"backend": {"host": "a", "port": 80, "proto": "http"}},
|
|
"/api": {"backend": {"host": "b", "port": 8080, "proto": "http"}},
|
|
},
|
|
}
|
|
}
|
|
domain_cfg = {"backend": "myapp"}
|
|
resolved = nginx._resolve_paths(domain_cfg, backends)
|
|
assert resolved == backends["myapp"]["paths"]
|
|
|
|
def test_returns_empty_when_backend_missing(self, temp_data_dir):
|
|
"""Empty dict when backend ref not found in backends."""
|
|
resolved = nginx._resolve_paths({"backend": "nonexistent"}, {})
|
|
assert resolved == {}
|
|
|
|
def test_returns_empty_when_no_backend_key(self, temp_data_dir):
|
|
"""Empty dict when domain has no backend key."""
|
|
resolved = nginx._resolve_paths({}, {"myapp": {"paths": {"/": {}}}})
|
|
assert resolved == {}
|
|
|
|
|
|
class TestResolveAuth:
|
|
def test_domain_auth_wins(self, temp_data_dir):
|
|
"""Domain-level auth overrides backend auth."""
|
|
backends = {
|
|
"myapp": {"auth": {"user": "backend", "htpasswd": "/backend/.htpasswd"}}
|
|
}
|
|
domain_cfg = {
|
|
"backend": "myapp",
|
|
"auth": {"user": "domain", "htpasswd": "/domain/.htpasswd"},
|
|
}
|
|
resolved = nginx._resolve_auth(domain_cfg, backends)
|
|
assert resolved == {"user": "domain", "htpasswd": "/domain/.htpasswd"}
|
|
|
|
def test_domain_auth_null_disables(self, temp_data_dir):
|
|
"""Domain auth set to None disables all auth."""
|
|
backends = {
|
|
"myapp": {"auth": {"user": "backend", "htpasswd": "/backend/.htpasswd"}}
|
|
}
|
|
domain_cfg = {"backend": "myapp", "auth": None}
|
|
resolved = nginx._resolve_auth(domain_cfg, backends)
|
|
assert resolved is None
|
|
|
|
def test_backend_auth_applies(self, temp_data_dir):
|
|
"""Backend auth applies when domain has no auth key."""
|
|
backends = {
|
|
"myapp": {"auth": {"user": "backend", "htpasswd": "/backend/.htpasswd"}}
|
|
}
|
|
domain_cfg = {"backend": "myapp"}
|
|
resolved = nginx._resolve_auth(domain_cfg, backends)
|
|
assert resolved == {"user": "backend", "htpasswd": "/backend/.htpasswd"}
|
|
|
|
def test_no_auth_when_absent_everywhere(self, temp_data_dir):
|
|
"""None when neither domain nor backend define auth."""
|
|
backends = {"myapp": {"paths": {}}}
|
|
domain_cfg = {"backend": "myapp"}
|
|
resolved = nginx._resolve_auth(domain_cfg, backends)
|
|
assert resolved is None
|
|
|
|
|
|
class TestBackendCRUD:
|
|
"""Test daemon handler backend CRUD operations directly."""
|
|
|
|
def test_validate_paths_valid(self, temp_data_dir):
|
|
"""_validate_paths accepts correct schemas."""
|
|
from daemon.handlers.nginx import _validate_paths
|
|
|
|
paths = {
|
|
"/": {"backend": {"host": "a", "port": 80, "proto": "http"}},
|
|
"/ws": {
|
|
"backend": {"host": "b", "port": 9091, "proto": "http"},
|
|
"is_websocket": True,
|
|
},
|
|
}
|
|
_validate_paths(paths) # no exception
|
|
|
|
def test_validate_paths_missing_host(self, temp_data_dir):
|
|
"""_validate_paths raises when backend.host missing."""
|
|
from daemon.handlers.nginx import _validate_paths
|
|
|
|
with pytest.raises(ValueError, match="missing 'host'"):
|
|
_validate_paths({"/": {"backend": {"port": 80, "proto": "http"}}})
|
|
|
|
def test_validate_paths_missing_port(self, temp_data_dir):
|
|
"""_validate_paths raises when backend.port missing."""
|
|
from daemon.handlers.nginx import _validate_paths
|
|
|
|
with pytest.raises(ValueError, match="missing 'port'"):
|
|
_validate_paths({"/": {"backend": {"host": "a", "proto": "http"}}})
|
|
|
|
def test_validate_paths_missing_proto(self, temp_data_dir):
|
|
"""_validate_paths raises when backend.proto missing."""
|
|
from daemon.handlers.nginx import _validate_paths
|
|
|
|
with pytest.raises(ValueError, match="missing 'proto'"):
|
|
_validate_paths({"/": {"backend": {"host": "a", "port": 80}}})
|
|
|
|
def test_validate_paths_no_backend(self, temp_data_dir):
|
|
"""_validate_paths raises when path lacks backend dict."""
|
|
from daemon.handlers.nginx import _validate_paths
|
|
|
|
with pytest.raises(ValueError, match="missing 'backend'"):
|
|
_validate_paths({"/": {"something": "else"}})
|
|
|
|
@patch("daemon.handlers.nginx._get_config")
|
|
@patch("daemon.handlers.nginx._save_config")
|
|
def test_add_backend(self, mock_save, mock_get, temp_data_dir):
|
|
"""_add_backend creates a new backend entry."""
|
|
from daemon.handlers.nginx import _add_backend
|
|
|
|
mock_get.return_value = {
|
|
"backends": {},
|
|
"domains": {},
|
|
"ssl": {},
|
|
}
|
|
|
|
_add_backend(
|
|
"test",
|
|
"Test Label",
|
|
{
|
|
"/": {"backend": {"host": "a", "port": 80, "proto": "http"}},
|
|
},
|
|
)
|
|
|
|
saved_cfg = mock_save.call_args[0][0]
|
|
assert "test" in saved_cfg["backends"]
|
|
assert saved_cfg["backends"]["test"]["label"] == "Test Label"
|
|
|
|
@patch("daemon.handlers.nginx._get_config")
|
|
@patch("daemon.handlers.nginx._save_config")
|
|
def test_add_backend_duplicate_raises(self, mock_save, mock_get, temp_data_dir):
|
|
"""_add_backend raises ValueError for duplicate name."""
|
|
from daemon.handlers.nginx import _add_backend
|
|
|
|
mock_get.return_value = {
|
|
"backends": {"test": {"label": "Existing"}},
|
|
"domains": {},
|
|
"ssl": {},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
_add_backend(
|
|
"test",
|
|
"New Label",
|
|
{
|
|
"/": {"backend": {"host": "a", "port": 80, "proto": "http"}},
|
|
},
|
|
)
|
|
|
|
@patch("daemon.handlers.nginx._get_config")
|
|
@patch("daemon.handlers.nginx._save_config")
|
|
def test_update_backend(self, mock_save, mock_get, temp_data_dir):
|
|
"""_update_backend modifies label and paths."""
|
|
from daemon.handlers.nginx import _update_backend
|
|
|
|
mock_get.return_value = {
|
|
"backends": {
|
|
"test": {
|
|
"label": "Old",
|
|
"paths": {
|
|
"/": {"backend": {"host": "a", "port": 80, "proto": "http"}}
|
|
},
|
|
}
|
|
},
|
|
"domains": {},
|
|
"ssl": {},
|
|
}
|
|
|
|
_update_backend(
|
|
"test",
|
|
label="New",
|
|
paths={
|
|
"/api": {"backend": {"host": "b", "port": 9000, "proto": "http"}},
|
|
},
|
|
)
|
|
|
|
saved_cfg = mock_save.call_args[0][0]
|
|
assert saved_cfg["backends"]["test"]["label"] == "New"
|
|
assert "/api" in saved_cfg["backends"]["test"]["paths"]
|
|
|
|
@patch("daemon.handlers.nginx._get_config")
|
|
def test_update_backend_notfound_raises(self, mock_get, temp_data_dir):
|
|
"""_update_backend raises KeyError for unknown backend."""
|
|
from daemon.handlers.nginx import _update_backend
|
|
|
|
mock_get.return_value = {"backends": {}, "domains": {}, "ssl": {}}
|
|
|
|
with pytest.raises(KeyError):
|
|
_update_backend("nonexistent", label="X")
|
|
|
|
@patch("daemon.handlers.nginx._get_config")
|
|
def test_update_backend_builtin_raises(self, mock_get, temp_data_dir):
|
|
"""_update_backend raises ValueError for builtin backend."""
|
|
from daemon.handlers.nginx import _update_backend
|
|
|
|
mock_get.return_value = {
|
|
"backends": {"webui": {"label": "WebUI", "builtin": True, "paths": {}}},
|
|
"domains": {},
|
|
"ssl": {},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="Cannot modify"):
|
|
_update_backend("webui", label="Hacked")
|
|
|
|
@patch("daemon.handlers.nginx._get_config")
|
|
@patch("daemon.handlers.nginx._save_config")
|
|
def test_remove_backend(self, mock_save, mock_get, temp_data_dir):
|
|
"""_remove_backend deletes non-builtin backend."""
|
|
from daemon.handlers.nginx import _remove_backend
|
|
|
|
mock_get.return_value = {
|
|
"backends": {"test": {"label": "Test", "paths": {}}},
|
|
"domains": {},
|
|
"ssl": {},
|
|
}
|
|
|
|
_remove_backend("test")
|
|
|
|
saved_cfg = mock_save.call_args[0][0]
|
|
assert "test" not in saved_cfg["backends"]
|
|
|
|
@patch("daemon.handlers.nginx._get_config")
|
|
def test_remove_backend_notfound_raises(self, mock_get, temp_data_dir):
|
|
"""_remove_backend raises KeyError for unknown backend."""
|
|
from daemon.handlers.nginx import _remove_backend
|
|
|
|
mock_get.return_value = {"backends": {}, "domains": {}, "ssl": {}}
|
|
|
|
with pytest.raises(KeyError):
|
|
_remove_backend("nonexistent")
|
|
|
|
@patch("daemon.handlers.nginx._get_config")
|
|
def test_remove_backend_builtin_raises(self, mock_get, temp_data_dir):
|
|
"""_remove_backend raises ValueError for builtin backend."""
|
|
from daemon.handlers.nginx import _remove_backend
|
|
|
|
mock_get.return_value = {
|
|
"backends": {"webui": {"label": "WebUI", "builtin": True, "paths": {}}},
|
|
"domains": {},
|
|
"ssl": {},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="Cannot remove"):
|
|
_remove_backend("webui")
|
|
|
|
@patch("daemon.handlers.nginx._get_config")
|
|
def test_remove_backend_referenced_raises_conflict(self, mock_get, temp_data_dir):
|
|
"""_remove_backend raises ConflictError when domains reference it."""
|
|
from daemon.handlers.nginx import _remove_backend
|
|
from daemon.server import ConflictError
|
|
|
|
mock_get.return_value = {
|
|
"backends": {"myapp": {"label": "App", "paths": {}}},
|
|
"domains": {"example.com": {"backend": "myapp"}},
|
|
"ssl": {},
|
|
}
|
|
|
|
with pytest.raises(ConflictError, match="referenced by domain"):
|
|
_remove_backend("myapp")
|
|
|
|
|
|
class TestMigration:
|
|
def test_ensure_webui_backend_creates(self, temp_data_dir):
|
|
"""_ensure_webui_backend creates webui backend if missing."""
|
|
cfg = {}
|
|
nginx._ensure_webui_backend(cfg)
|
|
assert "webui" in cfg["backends"]
|
|
assert cfg["backends"]["webui"]["_migrated"] is True
|
|
assert cfg["backends"]["webui"]["label"] == "Vacuum Wall WebUI"
|
|
assert cfg["backends"]["webui"]["builtin"] is True
|
|
|
|
def test_ensure_webui_backend_skips_migrated(self, temp_data_dir):
|
|
"""_ensure_webui_backend skips if _migrated is true."""
|
|
cfg = {
|
|
"backends": {
|
|
"webui": {
|
|
"label": "Custom",
|
|
"_migrated": True,
|
|
"paths": {},
|
|
}
|
|
}
|
|
}
|
|
nginx._ensure_webui_backend(cfg)
|
|
# Label unchanged — not recreated
|
|
assert cfg["backends"]["webui"]["label"] == "Custom"
|
|
|
|
def test_migrate_mgmt_domains_application_webui(self, temp_data_dir):
|
|
"""Domains with application='webui' and mgmt paths get backend='webui'."""
|
|
cfg = {
|
|
"backends": {"webui": {"_migrated": True, "paths": {}}},
|
|
"domains": {
|
|
"mgmt.example.com": {
|
|
"application": "webui",
|
|
"paths": {
|
|
"/": {"is_management": True},
|
|
"/ws": {"is_websocket": True},
|
|
},
|
|
"auth": {"user": "admin"},
|
|
}
|
|
},
|
|
}
|
|
nginx._migrate_mgmt_domains(cfg)
|
|
dom = cfg["domains"]["mgmt.example.com"]
|
|
assert dom["backend"] == "webui"
|
|
assert "application" not in dom
|
|
assert "paths" not in dom
|
|
assert "auth" not in dom
|
|
|
|
def test_migrate_mgmt_domains_strips_application_only(self, temp_data_dir):
|
|
"""application='webui' stripped even when paths don't match mgmt shape."""
|
|
cfg = {
|
|
"backends": {"webui": {"_migrated": True, "paths": {}}},
|
|
"domains": {
|
|
"odd.example.com": {
|
|
"application": "webui",
|
|
"paths": {"/": {"is_management": True}},
|
|
}
|
|
},
|
|
}
|
|
nginx._migrate_mgmt_domains(cfg)
|
|
dom = cfg["domains"]["odd.example.com"]
|
|
assert "application" not in dom
|
|
assert "backend" not in dom # paths didn't match full mgmt shape
|
|
assert "paths" in dom # not removed
|
|
|
|
def test_migrate_mgmt_domains_detects_webui_paths(self, temp_data_dir):
|
|
"""Domains matching webui path shape get migrated to backend."""
|
|
cfg = {
|
|
"backends": {"webui": {"_migrated": True, "paths": {}}},
|
|
"domains": {
|
|
"mgmt.example.com": {
|
|
"paths": {
|
|
"/": {
|
|
"backend": {
|
|
"host": "127.0.0.1",
|
|
"port": 9090,
|
|
"proto": "http",
|
|
},
|
|
"is_management": True,
|
|
},
|
|
"/ws": {
|
|
"backend": {
|
|
"host": "127.0.0.1",
|
|
"port": 9091,
|
|
"proto": "http",
|
|
},
|
|
"is_websocket": True,
|
|
},
|
|
}
|
|
}
|
|
},
|
|
}
|
|
nginx._migrate_mgmt_domains(cfg)
|
|
dom = cfg["domains"]["mgmt.example.com"]
|
|
assert dom["backend"] == "webui"
|
|
assert "paths" not in dom
|
|
|
|
def test_migrate_mgmt_domains_skips_non_mgmt(self, temp_data_dir):
|
|
"""Non-management domains are left untouched."""
|
|
cfg = {
|
|
"backends": {"webui": {"_migrated": True, "paths": {}}},
|
|
"domains": {
|
|
"app.example.com": {
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "10.0.0.1", "port": 80, "proto": "http"}
|
|
}
|
|
}
|
|
}
|
|
},
|
|
}
|
|
nginx._migrate_mgmt_domains(cfg)
|
|
dom = cfg["domains"]["app.example.com"]
|
|
assert "backend" not in dom
|
|
assert "paths" in dom
|
|
|
|
def test_migrate_config_full(self, temp_data_dir):
|
|
"""_migrate_config runs _ensure_webui_backend then _migrate_mgmt_domains."""
|
|
cfg = {
|
|
"domains": {
|
|
"mgmt.example.com": {
|
|
"application": "webui",
|
|
"paths": {
|
|
"/": {"is_management": True},
|
|
"/ws": {"is_websocket": True},
|
|
},
|
|
}
|
|
}
|
|
}
|
|
result = nginx._migrate_config(cfg)
|
|
assert result is cfg
|
|
assert "webui" in cfg["backends"]
|
|
assert cfg["domains"]["mgmt.example.com"]["backend"] == "webui"
|
|
|
|
|
|
class TestDomainSwap:
|
|
"""Domain can change which backend it references."""
|
|
|
|
def test_swap_backend(self, temp_data_dir):
|
|
"""update_domain allows changing backend field."""
|
|
cfg = {
|
|
"backends": {
|
|
"webui": {"label": "WebUI", "paths": {}},
|
|
"myapp": {"label": "App", "paths": {}},
|
|
},
|
|
"domains": {"example.com": {"backend": "webui", "force_ssl": True}},
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.save_config(cfg)
|
|
nginx.update_domain("example.com", backend="myapp")
|
|
loaded = nginx.get_config()
|
|
assert loaded["domains"]["example.com"]["backend"] == "myapp"
|
|
|
|
def test_swap_backend_nonexistent_raises(self, temp_data_dir):
|
|
"""update_domain raises ValueError when new backend not found."""
|
|
cfg = {
|
|
"backends": {"webui": {"label": "WebUI", "paths": {}}},
|
|
"domains": {"example.com": {"backend": "webui"}},
|
|
"ssl": {"protocols": "TLSv1.2 TLSv1.3"},
|
|
}
|
|
nginx.save_config(cfg)
|
|
with pytest.raises(ValueError, match="Backend 'nonexistent' not found"):
|
|
nginx.update_domain("example.com", backend="nonexistent")
|
|
|
|
|
|
class TestGetDomainsWithBackends:
|
|
"""get_domains resolves paths from backends, includes backend_name."""
|
|
|
|
def test_flattens_by_backend_paths(self, temp_data_dir):
|
|
"""Each backend path becomes a separate entry."""
|
|
cfg = {
|
|
"backends": {
|
|
"myapp": {
|
|
"label": "My App",
|
|
"paths": {
|
|
"/": {"backend": {"host": "a", "port": 80, "proto": "http"}},
|
|
"/api": {
|
|
"backend": {"host": "b", "port": 8080, "proto": "http"}
|
|
},
|
|
"/ws": {
|
|
"backend": {"host": "c", "port": 9091, "proto": "http"},
|
|
"is_websocket": True,
|
|
},
|
|
},
|
|
}
|
|
},
|
|
"domains": {"example.com": {"backend": "myapp", "force_ssl": True}},
|
|
"ssl": {},
|
|
}
|
|
nginx.save_config(cfg)
|
|
result = nginx.get_domains()
|
|
assert len(result) == 3
|
|
paths = {r["path"] for r in result}
|
|
assert paths == {"/", "/api", "/ws"}
|
|
for r in result:
|
|
assert r["backend_name"] == "myapp"
|
|
assert r["domain"] == "example.com"
|
|
|
|
def test_skips_domains_without_backend(self, temp_data_dir):
|
|
"""Domains without a backend key are skipped."""
|
|
cfg = {
|
|
"backends": {},
|
|
"domains": {
|
|
"good.com": {"backend": "webui"},
|
|
"bad.com": {"some": "orphan"},
|
|
},
|
|
"ssl": {},
|
|
}
|
|
nginx.save_config(cfg)
|
|
result = nginx.get_domains()
|
|
domains = {r["domain"] for r in result}
|
|
assert "good.com" in domains
|
|
assert "bad.com" not in domains
|
|
|
|
def test_includes_management_and_websocket_flags(self, temp_data_dir):
|
|
"""is_management and is_websocket flags propagate to entries."""
|
|
cfg = {
|
|
"backends": {
|
|
"webui": {
|
|
"label": "WebUI",
|
|
"paths": {
|
|
"/": {
|
|
"backend": {
|
|
"host": "127.0.0.1",
|
|
"port": 9090,
|
|
"proto": "http",
|
|
},
|
|
"is_management": True,
|
|
},
|
|
"/ws": {
|
|
"backend": {
|
|
"host": "127.0.0.1",
|
|
"port": 9091,
|
|
"proto": "http",
|
|
},
|
|
"is_websocket": True,
|
|
},
|
|
},
|
|
}
|
|
},
|
|
"domains": {"mgmt.local": {"backend": "webui"}},
|
|
"ssl": {},
|
|
}
|
|
nginx.save_config(cfg)
|
|
result = nginx.get_domains()
|
|
entries_by_path = {r["path"]: r for r in result}
|
|
assert entries_by_path["/"]["is_management"] is True
|
|
assert entries_by_path["/ws"]["is_websocket"] is True
|
|
|
|
|
|
class TestGenerateServerConfWithBackends:
|
|
"""generate_server_conf resolves paths via backends parameter."""
|
|
|
|
def test_with_backends_param(self, temp_data_dir):
|
|
"""When backends provided, paths resolve from backend."""
|
|
backends = {
|
|
"myapp": {
|
|
"paths": {
|
|
"/": {
|
|
"backend": {"host": "10.0.0.1", "port": 8080, "proto": "http"}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
domain_cfg = {"domain": "example.com", "backend": "myapp", "force_ssl": True}
|
|
out = nginx.generate_server_conf(domain_cfg, backends)
|
|
assert "proxy_pass http://10.0.0.1:8080;" in out
|
|
|
|
|
|
class TestPatchConfigPreservesBackends:
|
|
"""PATCH_NGINX_CONFIG (deep_merge) does not overwrite backends fully."""
|
|
|
|
def test_ssl_patch_keeps_backends(self, temp_data_dir):
|
|
"""Patching ssl settings preserves backends dict (deep_merge semantics)."""
|
|
from lib.common import deep_merge
|
|
|
|
current = {
|
|
"backends": {"myapp": {"label": "App", "paths": {}}},
|
|
"domains": {"example.com": {"backend": "myapp"}},
|
|
"ssl": {"protocols": "TLSv1.2"},
|
|
}
|
|
patch = {"ssl": {"protocols": "TLSv1.3"}}
|
|
merged = deep_merge(current, patch)
|
|
assert merged["backends"]["myapp"]["label"] == "App"
|
|
assert merged["domains"]["example.com"]["backend"] == "myapp"
|
|
assert merged["ssl"]["protocols"] == "TLSv1.3"
|