removed .tests, not needed
This commit is contained in:
@@ -1,166 +0,0 @@
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import types
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
|
||||
def _install_web_stubs():
|
||||
if "fastapi" in sys.modules:
|
||||
return
|
||||
|
||||
class _HTTPException(Exception):
|
||||
def __init__(self, status_code=None, detail=None):
|
||||
super().__init__(detail)
|
||||
self.status_code = status_code
|
||||
self.detail = detail
|
||||
|
||||
class _FastAPI:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
def mount(self, *args, **kwargs):
|
||||
return None
|
||||
|
||||
def add_middleware(self, *args, **kwargs):
|
||||
return None
|
||||
|
||||
def __getattr__(self, _name):
|
||||
def _decorator_factory(*args, **kwargs):
|
||||
def _decorator(func):
|
||||
return func
|
||||
|
||||
return _decorator
|
||||
|
||||
return _decorator_factory
|
||||
|
||||
class _BaseModel:
|
||||
pass
|
||||
|
||||
class _StaticFiles:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
class _Jinja2Templates:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
class _BaseHTTPMiddleware:
|
||||
pass
|
||||
|
||||
fastapi_module = types.ModuleType("fastapi")
|
||||
fastapi_module.FastAPI = _FastAPI
|
||||
fastapi_module.HTTPException = _HTTPException
|
||||
sys.modules["fastapi"] = fastapi_module
|
||||
|
||||
responses_module = types.ModuleType("fastapi.responses")
|
||||
responses_module.HTMLResponse = object
|
||||
responses_module.JSONResponse = object
|
||||
responses_module.RedirectResponse = object
|
||||
sys.modules["fastapi.responses"] = responses_module
|
||||
|
||||
staticfiles_module = types.ModuleType("fastapi.staticfiles")
|
||||
staticfiles_module.StaticFiles = _StaticFiles
|
||||
sys.modules["fastapi.staticfiles"] = staticfiles_module
|
||||
|
||||
templating_module = types.ModuleType("fastapi.templating")
|
||||
templating_module.Jinja2Templates = _Jinja2Templates
|
||||
sys.modules["fastapi.templating"] = templating_module
|
||||
|
||||
requests_module = types.ModuleType("fastapi.requests")
|
||||
requests_module.Request = object
|
||||
sys.modules["fastapi.requests"] = requests_module
|
||||
|
||||
pydantic_module = types.ModuleType("pydantic")
|
||||
pydantic_module.BaseModel = _BaseModel
|
||||
sys.modules["pydantic"] = pydantic_module
|
||||
|
||||
starlette_base_module = types.ModuleType("starlette.middleware.base")
|
||||
starlette_base_module.BaseHTTPMiddleware = _BaseHTTPMiddleware
|
||||
sys.modules["starlette.middleware.base"] = starlette_base_module
|
||||
|
||||
starlette_middleware_module = types.ModuleType("starlette.middleware")
|
||||
starlette_middleware_module.base = starlette_base_module
|
||||
sys.modules["starlette.middleware"] = starlette_middleware_module
|
||||
|
||||
starlette_module = types.ModuleType("starlette")
|
||||
starlette_module.middleware = starlette_middleware_module
|
||||
sys.modules["starlette"] = starlette_module
|
||||
|
||||
|
||||
_install_web_stubs()
|
||||
from sovran_systemsos_web import server
|
||||
|
||||
|
||||
class Bip110StatusTests(unittest.TestCase):
|
||||
def _status(self, deploy_info, net_info):
|
||||
with patch.object(server, "_get_bitcoin_deployment_info", return_value=deploy_info), patch.object(
|
||||
server, "_get_bitcoin_version_info", return_value=net_info
|
||||
):
|
||||
return server._get_bip110_status()
|
||||
|
||||
def test_started_reduced_data_reports_signaling(self):
|
||||
deploy_info = {
|
||||
"deployments": {
|
||||
"reduced_data": {
|
||||
"type": "bip9",
|
||||
"active": False,
|
||||
"bip9": {
|
||||
"bit": 4,
|
||||
"status": "started",
|
||||
"statistics": {"elapsed": 833, "count": 4, "threshold": 1109},
|
||||
"signalling": "--#--",
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result = self._status(deploy_info, {"subversion": "/Satoshi:29.0.0/"})
|
||||
self.assertEqual(
|
||||
result,
|
||||
{"supported": True, "signaling": True, "state": "signaling", "source": "getdeploymentinfo"},
|
||||
)
|
||||
|
||||
def test_active_reduced_data_reports_active(self):
|
||||
deploy_info = {
|
||||
"deployments": {"reduced_data": {"active": True, "bip9": {"bit": 4, "status": "active"}}}
|
||||
}
|
||||
|
||||
result = self._status(deploy_info, {"subversion": "/Satoshi:29.0.0/"})
|
||||
self.assertEqual(result["state"], "active")
|
||||
self.assertTrue(result["supported"])
|
||||
self.assertTrue(result["signaling"])
|
||||
self.assertEqual(result["source"], "getdeploymentinfo")
|
||||
|
||||
def test_locked_in_reduced_data_reports_locked_in(self):
|
||||
deploy_info = {
|
||||
"deployments": {"reduced_data": {"active": False, "bip9": {"bit": 4, "status": "locked_in"}}}
|
||||
}
|
||||
|
||||
result = self._status(deploy_info, {"subversion": "/Satoshi:29.0.0/"})
|
||||
self.assertEqual(result["state"], "locked_in")
|
||||
self.assertTrue(result["supported"])
|
||||
self.assertTrue(result["signaling"])
|
||||
self.assertEqual(result["source"], "getdeploymentinfo")
|
||||
|
||||
def test_no_bip110_deployment_and_plain_subversion_reports_unsupported(self):
|
||||
deploy_info = {
|
||||
"deployments": {
|
||||
"taproot": {"type": "bip9", "active": True, "bip9": {"bit": 2, "status": "active"}},
|
||||
}
|
||||
}
|
||||
result = self._status(deploy_info, {"subversion": "/Satoshi:27.0.0/"})
|
||||
self.assertEqual(
|
||||
result,
|
||||
{"supported": False, "signaling": False, "state": "unsupported", "source": "subversion"},
|
||||
)
|
||||
|
||||
def test_node_unreachable_reports_unknown(self):
|
||||
result = self._status(None, None)
|
||||
self.assertEqual(result, {"supported": False, "signaling": False, "state": "unknown", "source": "none"})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,509 +0,0 @@
|
||||
"""
|
||||
Domain conflict and unique-hostname tests.
|
||||
|
||||
Tests cover:
|
||||
1. Wallet Connections initial setup displays unique-hostname guidance.
|
||||
2. Wallet Connections reconfiguration displays the same guidance.
|
||||
3. The field example is lightning.yourdomain.com for Wallet Connections.
|
||||
4. An unused hostname such as lightning.example.com is accepted.
|
||||
5. Reusing a hostname assigned to Matrix, Nextcloud, WordPress, BTCPay Server,
|
||||
Vaultwarden, Haven, or Element Calling returns HTTP 409.
|
||||
6. Comparison is case-insensitive.
|
||||
7. A hostname with one trailing dot conflicts with the equivalent hostname without it.
|
||||
8. Re-saving the existing lightning hostname for lightning remains allowed.
|
||||
9. Invalid hostnames are rejected before mutation.
|
||||
10. On conflict, the domain file remains unchanged.
|
||||
11. On conflict, the DDNS script remains unchanged and is not executed.
|
||||
12. Generic domain flows for unrelated services remain intact (matrix -> matrix, etc.).
|
||||
13. JavaScript syntax checks pass for features.js and helpers.js.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
|
||||
# ── Minimal stubs so server.py can be imported without full FastAPI ──
|
||||
|
||||
def _install_web_stubs():
|
||||
if "fastapi" in sys.modules:
|
||||
return
|
||||
|
||||
class _HTTPException(Exception):
|
||||
def __init__(self, status_code=None, detail=None):
|
||||
super().__init__(detail)
|
||||
self.status_code = status_code
|
||||
self.detail = detail
|
||||
|
||||
class _FastAPI:
|
||||
def __init__(self, *a, **kw): pass
|
||||
def mount(self, *a, **kw): return None
|
||||
def add_middleware(self, *a, **kw): return None
|
||||
def __getattr__(self, _name):
|
||||
def _deco_factory(*a, **kw):
|
||||
def _deco(func): return func
|
||||
return _deco
|
||||
return _deco_factory
|
||||
|
||||
class _BaseModel: pass
|
||||
|
||||
class _JSONResponse:
|
||||
def __init__(self, content=None, status_code=200):
|
||||
self.content = content
|
||||
self.status_code = status_code
|
||||
self.body = json.dumps(content or {}).encode("utf-8")
|
||||
|
||||
fastapi_mod = types.ModuleType("fastapi")
|
||||
fastapi_mod.FastAPI = _FastAPI
|
||||
fastapi_mod.HTTPException = _HTTPException
|
||||
sys.modules["fastapi"] = fastapi_mod
|
||||
|
||||
resp_mod = types.ModuleType("fastapi.responses")
|
||||
resp_mod.HTMLResponse = object
|
||||
resp_mod.RedirectResponse = object
|
||||
resp_mod.JSONResponse = _JSONResponse
|
||||
sys.modules["fastapi.responses"] = resp_mod
|
||||
|
||||
sys.modules["fastapi.staticfiles"] = types.ModuleType("fastapi.staticfiles")
|
||||
|
||||
class _StaticFiles:
|
||||
def __init__(self, *args, **kwargs): pass
|
||||
|
||||
sys.modules["fastapi.staticfiles"].StaticFiles = _StaticFiles
|
||||
|
||||
class _Jinja2Templates:
|
||||
def __init__(self, *args, **kwargs): pass
|
||||
|
||||
tmpl_mod = types.ModuleType("fastapi.templating")
|
||||
tmpl_mod.Jinja2Templates = _Jinja2Templates
|
||||
sys.modules["fastapi.templating"] = tmpl_mod
|
||||
|
||||
req_mod = types.ModuleType("fastapi.requests")
|
||||
req_mod.Request = object
|
||||
sys.modules["fastapi.requests"] = req_mod
|
||||
|
||||
pyd_mod = types.ModuleType("pydantic")
|
||||
pyd_mod.BaseModel = _BaseModel
|
||||
sys.modules["pydantic"] = pyd_mod
|
||||
|
||||
stl_base = types.ModuleType("starlette.middleware.base")
|
||||
stl_base.BaseHTTPMiddleware = object
|
||||
sys.modules["starlette.middleware.base"] = stl_base
|
||||
stl_mw = types.ModuleType("starlette.middleware")
|
||||
sys.modules["starlette.middleware"] = stl_mw
|
||||
stl = types.ModuleType("starlette")
|
||||
sys.modules["starlette"] = stl
|
||||
|
||||
|
||||
_install_web_stubs()
|
||||
|
||||
from sovran_systemsos_web import server # noqa: E402
|
||||
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────
|
||||
|
||||
def _make_req(domain_name, domain, ddns_url=""):
|
||||
"""Build a DomainSetRequest-like object using the server's model."""
|
||||
req = object.__new__(server.DomainSetRequest)
|
||||
req.domain_name = domain_name
|
||||
req.domain = domain
|
||||
req.ddns_url = ddns_url
|
||||
return req
|
||||
|
||||
|
||||
def _write_domain_file(domains_dir, key, value):
|
||||
path = os.path.join(domains_dir, key)
|
||||
with open(path, "w") as fh:
|
||||
fh.write(value)
|
||||
|
||||
|
||||
# ── Unit tests for server helper functions ─────────────────────────
|
||||
|
||||
class NormalizeHostnameTests(unittest.TestCase):
|
||||
def test_trims_whitespace(self):
|
||||
self.assertEqual(server._normalize_hostname(" foo.example.com "), "foo.example.com")
|
||||
|
||||
def test_lowercases(self):
|
||||
self.assertEqual(server._normalize_hostname("FOO.Example.COM"), "foo.example.com")
|
||||
|
||||
def test_removes_exactly_one_trailing_dot(self):
|
||||
self.assertEqual(server._normalize_hostname("foo.example.com."), "foo.example.com")
|
||||
|
||||
def test_does_not_remove_two_trailing_dots(self):
|
||||
# Only one trailing dot is removed; two trailing dots leave one.
|
||||
self.assertEqual(server._normalize_hostname("foo.example.com.."), "foo.example.com.")
|
||||
|
||||
def test_no_trailing_dot_unchanged(self):
|
||||
self.assertEqual(server._normalize_hostname("foo.example.com"), "foo.example.com")
|
||||
|
||||
def test_strips_and_lowercases_with_trailing_dot(self):
|
||||
self.assertEqual(server._normalize_hostname(" Lightning.Example.COM. "), "lightning.example.com")
|
||||
|
||||
|
||||
class ValidateHostnameTests(unittest.TestCase):
|
||||
def test_valid_simple_domain(self):
|
||||
self.assertTrue(server._validate_hostname("foo.example.com"))
|
||||
|
||||
def test_valid_subdomain(self):
|
||||
self.assertTrue(server._validate_hostname("lightning.yourdomain.com"))
|
||||
|
||||
def test_valid_bare_hostname(self):
|
||||
self.assertTrue(server._validate_hostname("example"))
|
||||
|
||||
def test_valid_with_hyphens(self):
|
||||
self.assertTrue(server._validate_hostname("my-host.example.com"))
|
||||
|
||||
def test_invalid_empty(self):
|
||||
self.assertFalse(server._validate_hostname(""))
|
||||
|
||||
def test_invalid_trailing_dot(self):
|
||||
# The hostname validator rejects trailing dots directly.
|
||||
# In the API flow, normalization removes exactly one trailing dot before
|
||||
# this validator is called, so a single trailing dot in user input is
|
||||
# handled before reaching validation.
|
||||
self.assertFalse(server._validate_hostname("foo.example.com."))
|
||||
|
||||
def test_invalid_with_underscore(self):
|
||||
self.assertFalse(server._validate_hostname("foo_bar.example.com"))
|
||||
|
||||
def test_invalid_leading_hyphen(self):
|
||||
self.assertFalse(server._validate_hostname("-foo.example.com"))
|
||||
|
||||
def test_invalid_spaces(self):
|
||||
self.assertFalse(server._validate_hostname("foo example.com"))
|
||||
|
||||
|
||||
class CheckDomainConflictTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmpdir = tempfile.mkdtemp()
|
||||
self._orig_domains_dir = server.DOMAINS_DIR
|
||||
server.DOMAINS_DIR = self.tmpdir
|
||||
|
||||
def tearDown(self):
|
||||
server.DOMAINS_DIR = self._orig_domains_dir
|
||||
import shutil
|
||||
shutil.rmtree(self.tmpdir, ignore_errors=True)
|
||||
|
||||
def test_no_conflict_when_no_other_files(self):
|
||||
result = server._check_domain_conflict("lightning", "lightning.example.com")
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_conflict_when_matrix_has_same_hostname(self):
|
||||
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
|
||||
result = server._check_domain_conflict("lightning", "shared.example.com")
|
||||
self.assertEqual(result, "matrix")
|
||||
|
||||
def test_conflict_when_nextcloud_has_same_hostname(self):
|
||||
_write_domain_file(self.tmpdir, "nextcloud", "shared.example.com")
|
||||
result = server._check_domain_conflict("lightning", "shared.example.com")
|
||||
self.assertEqual(result, "nextcloud")
|
||||
|
||||
def test_conflict_when_wordpress_has_same_hostname(self):
|
||||
_write_domain_file(self.tmpdir, "wordpress", "shared.example.com")
|
||||
result = server._check_domain_conflict("lightning", "shared.example.com")
|
||||
self.assertEqual(result, "wordpress")
|
||||
|
||||
def test_conflict_when_btcpayserver_has_same_hostname(self):
|
||||
_write_domain_file(self.tmpdir, "btcpayserver", "shared.example.com")
|
||||
result = server._check_domain_conflict("lightning", "shared.example.com")
|
||||
self.assertEqual(result, "btcpayserver")
|
||||
|
||||
def test_conflict_when_vaultwarden_has_same_hostname(self):
|
||||
_write_domain_file(self.tmpdir, "vaultwarden", "shared.example.com")
|
||||
result = server._check_domain_conflict("lightning", "shared.example.com")
|
||||
self.assertEqual(result, "vaultwarden")
|
||||
|
||||
def test_conflict_when_haven_has_same_hostname(self):
|
||||
_write_domain_file(self.tmpdir, "haven", "shared.example.com")
|
||||
result = server._check_domain_conflict("lightning", "shared.example.com")
|
||||
self.assertEqual(result, "haven")
|
||||
|
||||
def test_conflict_when_element_calling_has_same_hostname(self):
|
||||
_write_domain_file(self.tmpdir, "element-calling", "shared.example.com")
|
||||
result = server._check_domain_conflict("lightning", "shared.example.com")
|
||||
self.assertEqual(result, "element-calling")
|
||||
|
||||
def test_no_conflict_for_self(self):
|
||||
# Re-saving lightning's own existing hostname must be allowed.
|
||||
_write_domain_file(self.tmpdir, "lightning", "lightning.example.com")
|
||||
result = server._check_domain_conflict("lightning", "lightning.example.com")
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_symmetric_conflict_matrix_reusing_lightning(self):
|
||||
# Saving matrix with lightning's existing hostname is also rejected.
|
||||
_write_domain_file(self.tmpdir, "lightning", "shared.example.com")
|
||||
result = server._check_domain_conflict("matrix", "shared.example.com")
|
||||
self.assertEqual(result, "lightning")
|
||||
|
||||
def test_no_conflict_unrelated_services_without_lightning(self):
|
||||
# Two unrelated non-lightning services with the same hostname.
|
||||
# The rule only applies when lightning is involved.
|
||||
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
|
||||
result = server._check_domain_conflict("nextcloud", "shared.example.com")
|
||||
self.assertIsNone(result)
|
||||
|
||||
|
||||
# ── API endpoint integration-style tests ──────────────────────────
|
||||
|
||||
import asyncio
|
||||
|
||||
|
||||
def _run(coro):
|
||||
"""Run an async coroutine synchronously."""
|
||||
return asyncio.get_event_loop().run_until_complete(coro)
|
||||
|
||||
|
||||
class ApiDomainsSetConflictTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmpdir = tempfile.mkdtemp()
|
||||
self.njalla_dir = tempfile.mkdtemp()
|
||||
self.njalla_script = os.path.join(self.njalla_dir, "njalla.sh")
|
||||
self._orig_domains_dir = server.DOMAINS_DIR
|
||||
self._orig_njalla = server.NJALLA_SCRIPT
|
||||
server.DOMAINS_DIR = self.tmpdir
|
||||
server.NJALLA_SCRIPT = self.njalla_script
|
||||
|
||||
def tearDown(self):
|
||||
server.DOMAINS_DIR = self._orig_domains_dir
|
||||
server.NJALLA_SCRIPT = self._orig_njalla
|
||||
import shutil
|
||||
shutil.rmtree(self.tmpdir, ignore_errors=True)
|
||||
shutil.rmtree(self.njalla_dir, ignore_errors=True)
|
||||
|
||||
# ── Test 4: unused hostname is accepted ──────────────────────────
|
||||
def test_unused_hostname_is_accepted(self):
|
||||
with patch.object(server, "_trigger_hosts_update"):
|
||||
result = _run(server.api_domains_set(_make_req("lightning", "lightning.example.com")))
|
||||
self.assertEqual(result.get("ok"), True)
|
||||
# Domain file should be written.
|
||||
with open(os.path.join(self.tmpdir, "lightning")) as fh:
|
||||
saved = fh.read()
|
||||
self.assertEqual(saved, "lightning.example.com")
|
||||
|
||||
# ── Test 5: conflict with each managed service returns 409 ────────
|
||||
def _assert_conflict_409(self, target_key, conflicting_key, hostname):
|
||||
_write_domain_file(self.tmpdir, conflicting_key, hostname)
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
_run(server.api_domains_set(_make_req(target_key, hostname)))
|
||||
exc = ctx.exception
|
||||
self.assertEqual(getattr(exc, "status_code", None), 409)
|
||||
detail = getattr(exc, "detail", {})
|
||||
self.assertEqual(detail.get("error"), "domain_conflict")
|
||||
self.assertEqual(detail.get("conflicting_domain_key"), conflicting_key)
|
||||
self.assertIn("message", detail)
|
||||
|
||||
def test_conflict_with_matrix_returns_409(self):
|
||||
self._assert_conflict_409("lightning", "matrix", "shared.example.com")
|
||||
|
||||
def test_conflict_with_nextcloud_returns_409(self):
|
||||
self._assert_conflict_409("lightning", "nextcloud", "shared.example.com")
|
||||
|
||||
def test_conflict_with_wordpress_returns_409(self):
|
||||
self._assert_conflict_409("lightning", "wordpress", "shared.example.com")
|
||||
|
||||
def test_conflict_with_btcpayserver_returns_409(self):
|
||||
self._assert_conflict_409("lightning", "btcpayserver", "shared.example.com")
|
||||
|
||||
def test_conflict_with_vaultwarden_returns_409(self):
|
||||
self._assert_conflict_409("lightning", "vaultwarden", "shared.example.com")
|
||||
|
||||
def test_conflict_with_haven_returns_409(self):
|
||||
self._assert_conflict_409("lightning", "haven", "shared.example.com")
|
||||
|
||||
def test_conflict_with_element_calling_returns_409(self):
|
||||
self._assert_conflict_409("lightning", "element-calling", "shared.example.com")
|
||||
|
||||
# ── Test 6: comparison is case-insensitive ───────────────────────
|
||||
def test_conflict_is_case_insensitive(self):
|
||||
_write_domain_file(self.tmpdir, "matrix", "Shared.Example.COM")
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
_run(server.api_domains_set(_make_req("lightning", "shared.example.com")))
|
||||
self.assertEqual(getattr(ctx.exception, "status_code", None), 409)
|
||||
|
||||
def test_conflict_case_insensitive_reversed(self):
|
||||
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
_run(server.api_domains_set(_make_req("lightning", "SHARED.EXAMPLE.COM")))
|
||||
self.assertEqual(getattr(ctx.exception, "status_code", None), 409)
|
||||
|
||||
# ── Test 7: trailing-dot normalization causes conflict ────────────
|
||||
def test_trailing_dot_conflicts_with_same_hostname(self):
|
||||
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
_run(server.api_domains_set(_make_req("lightning", "shared.example.com.")))
|
||||
self.assertEqual(getattr(ctx.exception, "status_code", None), 409)
|
||||
|
||||
def test_stored_trailing_dot_conflicts_with_clean_submission(self):
|
||||
_write_domain_file(self.tmpdir, "matrix", "shared.example.com.")
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
_run(server.api_domains_set(_make_req("lightning", "shared.example.com")))
|
||||
self.assertEqual(getattr(ctx.exception, "status_code", None), 409)
|
||||
|
||||
# ── Test 8: re-saving existing lightning hostname is allowed ──────
|
||||
def test_resave_existing_lightning_hostname_allowed(self):
|
||||
_write_domain_file(self.tmpdir, "lightning", "lightning.example.com")
|
||||
with patch.object(server, "_trigger_hosts_update"):
|
||||
result = _run(server.api_domains_set(_make_req("lightning", "lightning.example.com")))
|
||||
self.assertEqual(result.get("ok"), True)
|
||||
|
||||
# ── Test 9: invalid hostname is rejected before mutation ──────────
|
||||
def test_invalid_hostname_rejected(self):
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
_run(server.api_domains_set(_make_req("lightning", "not a hostname!")))
|
||||
self.assertEqual(getattr(ctx.exception, "status_code", None), 400)
|
||||
|
||||
def test_hostname_with_underscore_rejected(self):
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
_run(server.api_domains_set(_make_req("lightning", "foo_bar.example.com")))
|
||||
self.assertEqual(getattr(ctx.exception, "status_code", None), 400)
|
||||
|
||||
# ── Test 10: on conflict, domain file remains unchanged ───────────
|
||||
def test_domain_file_unchanged_on_conflict(self):
|
||||
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
|
||||
lightning_path = os.path.join(self.tmpdir, "lightning")
|
||||
# Write a prior value.
|
||||
_write_domain_file(self.tmpdir, "lightning", "old.example.com")
|
||||
with self.assertRaises(Exception):
|
||||
_run(server.api_domains_set(_make_req("lightning", "shared.example.com")))
|
||||
# The lightning domain file must still contain the old value.
|
||||
with open(lightning_path) as fh:
|
||||
saved = fh.read()
|
||||
self.assertEqual(saved, "old.example.com")
|
||||
|
||||
def test_domain_file_not_created_on_conflict_if_absent(self):
|
||||
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
|
||||
lightning_path = os.path.join(self.tmpdir, "lightning")
|
||||
self.assertFalse(os.path.exists(lightning_path))
|
||||
with self.assertRaises(Exception):
|
||||
_run(server.api_domains_set(_make_req("lightning", "shared.example.com")))
|
||||
self.assertFalse(os.path.exists(lightning_path))
|
||||
|
||||
# ── Test 11: on conflict, DDNS script unchanged and not executed ──
|
||||
def test_njalla_script_not_written_on_conflict(self):
|
||||
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
|
||||
with self.assertRaises(Exception):
|
||||
_run(server.api_domains_set(
|
||||
_make_req("lightning", "shared.example.com",
|
||||
ddns_url='curl "https://njal.la/update/?h=shared.example.com&k=key&auto"')
|
||||
))
|
||||
self.assertFalse(os.path.exists(self.njalla_script))
|
||||
|
||||
def test_njalla_script_not_executed_on_conflict(self):
|
||||
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
|
||||
with patch("subprocess.run") as mock_run, self.assertRaises(Exception):
|
||||
_run(server.api_domains_set(
|
||||
_make_req("lightning", "shared.example.com",
|
||||
ddns_url='curl "https://njal.la/update/?h=shared.example.com&k=key&auto"')
|
||||
))
|
||||
mock_run.assert_not_called()
|
||||
|
||||
# ── Test 12: generic domain flows for unrelated services intact ───
|
||||
def test_matrix_save_without_lightning_succeeds(self):
|
||||
with patch.object(server, "_trigger_hosts_update"):
|
||||
result = _run(server.api_domains_set(_make_req("matrix", "matrix.example.com")))
|
||||
self.assertEqual(result.get("ok"), True)
|
||||
with open(os.path.join(self.tmpdir, "matrix")) as fh:
|
||||
saved = fh.read()
|
||||
self.assertEqual(saved, "matrix.example.com")
|
||||
|
||||
def test_nextcloud_save_without_lightning_succeeds(self):
|
||||
with patch.object(server, "_trigger_hosts_update"):
|
||||
result = _run(server.api_domains_set(_make_req("nextcloud", "cloud.example.com")))
|
||||
self.assertEqual(result.get("ok"), True)
|
||||
|
||||
def test_two_non_lightning_services_sharing_hostname_allowed(self):
|
||||
# The uniqueness rule is only enforced when lightning is involved.
|
||||
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
|
||||
with patch.object(server, "_trigger_hosts_update"):
|
||||
result = _run(server.api_domains_set(_make_req("nextcloud", "shared.example.com")))
|
||||
self.assertEqual(result.get("ok"), True)
|
||||
|
||||
|
||||
# ── Test 13: JavaScript syntax checks ─────────────────────────────
|
||||
|
||||
class JsSyntaxTests(unittest.TestCase):
|
||||
def _js_files(self):
|
||||
static_js = Path(__file__).resolve().parents[1] / "sovran_systemsos_web" / "static" / "js"
|
||||
return sorted(static_js.glob("*.js"))
|
||||
|
||||
def test_js_syntax_no_errors(self):
|
||||
for js_file in self._js_files():
|
||||
with self.subTest(file=js_file.name):
|
||||
result = subprocess.run(
|
||||
["node", "--check", str(js_file)],
|
||||
capture_output=True, text=True
|
||||
)
|
||||
self.assertEqual(
|
||||
result.returncode, 0,
|
||||
msg=f"Syntax error in {js_file.name}:\n{result.stderr}"
|
||||
)
|
||||
|
||||
|
||||
# ── UI guidance content tests (features.js) ───────────────────────
|
||||
|
||||
class FeaturesJsContentTests(unittest.TestCase):
|
||||
"""Verify the features.js source contains the required Wallet Connections guidance."""
|
||||
|
||||
def setUp(self):
|
||||
self.features_js = (
|
||||
Path(__file__).resolve().parents[1]
|
||||
/ "sovran_systemsos_web" / "static" / "js" / "features.js"
|
||||
).read_text(encoding="utf-8")
|
||||
|
||||
# ── Test 1: initial setup displays unique-hostname guidance ───────
|
||||
def test_setup_modal_contains_nwc_warning(self):
|
||||
self.assertIn("Wallet Connections requires its own unique hostname", self.features_js)
|
||||
|
||||
# ── Test 2: reconfiguration displays the same guidance ────────────
|
||||
def test_reconfig_modal_contains_nwc_warning(self):
|
||||
# The warning text must appear in both openDomainSetupModal and
|
||||
# openDomainReconfigureModal — two occurrences minimum.
|
||||
count = self.features_js.count("Wallet Connections requires its own unique hostname")
|
||||
self.assertGreaterEqual(count, 2, "Warning must appear in both setup and reconfigure modals")
|
||||
|
||||
# ── Test 3: field example is lightning.yourdomain.com ────────────
|
||||
def test_lightning_placeholder_is_present(self):
|
||||
self.assertIn("lightning.yourdomain.com", self.features_js)
|
||||
|
||||
# ── Warning references correct services ───────────────────────────
|
||||
def test_warning_mentions_matrix(self):
|
||||
self.assertIn("Matrix", self.features_js)
|
||||
|
||||
def test_warning_mentions_nextcloud(self):
|
||||
self.assertIn("Nextcloud", self.features_js)
|
||||
|
||||
def test_warning_mentions_btcpay_server(self):
|
||||
self.assertIn("BTCPay Server", self.features_js)
|
||||
|
||||
def test_warning_mentions_vaultwarden(self):
|
||||
self.assertIn("Vaultwarden", self.features_js)
|
||||
|
||||
def test_warning_mentions_haven(self):
|
||||
self.assertIn("Haven", self.features_js)
|
||||
|
||||
def test_warning_mentions_wordpress(self):
|
||||
self.assertIn("WordPress", self.features_js)
|
||||
|
||||
# ── isWalletConnections guard targets correct identifiers ─────────
|
||||
def test_nwc_wallets_id_check_present(self):
|
||||
self.assertIn('feat.id === "nwc-wallets"', self.features_js)
|
||||
|
||||
def test_lightning_domain_name_check_present(self):
|
||||
self.assertIn('feat.domain_name === "lightning"', self.features_js)
|
||||
|
||||
# ── Error surfacing ───────────────────────────────────────────────
|
||||
def test_error_message_surfaced_in_setup_modal(self):
|
||||
# The catch block must use err.message rather than hard-coded string.
|
||||
self.assertIn("err.message", self.features_js)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,44 +0,0 @@
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
HUB_NIX = Path(__file__).resolve().parents[2] / "modules" / "core" / "sovran-hub.nix"
|
||||
|
||||
|
||||
def _section(source: str, start: str, end: str) -> str:
|
||||
start_idx = source.find(start)
|
||||
if start_idx == -1:
|
||||
raise AssertionError(f"Expected section start not found: {start!r}")
|
||||
end_idx = source.find(end, start_idx)
|
||||
if end_idx == -1:
|
||||
raise AssertionError(f"Expected section end not found: {end!r}")
|
||||
return source[start_idx:end_idx]
|
||||
|
||||
|
||||
class HubUpdateBootStagingTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.source = HUB_NIX.read_text()
|
||||
self.update_section = _section(
|
||||
self.source,
|
||||
'update-script = pkgs.writeShellScript "sovran-hub-update.sh" \'\'',
|
||||
"# ── Rebuild wrapper script",
|
||||
)
|
||||
self.rebuild_section = _section(
|
||||
self.source,
|
||||
'rebuild-script = pkgs.writeShellScript "sovran-hub-rebuild.sh" \'\'',
|
||||
"# ── Brave launcher wrapper",
|
||||
)
|
||||
|
||||
def test_full_update_uses_boot_not_switch(self):
|
||||
self.assertIn("nixos-rebuild boot --flake /etc/nixos", self.update_section)
|
||||
self.assertNotIn("nixos-rebuild switch --flake /etc/nixos", self.update_section)
|
||||
|
||||
def test_full_update_marks_reboot_required(self):
|
||||
self.assertIn('echo "REBOOT_REQUIRED" > "$STATUS"', self.update_section)
|
||||
|
||||
def test_rebuild_path_keeps_switch_semantics(self):
|
||||
self.assertIn("nixos-rebuild switch --flake /etc/nixos", self.rebuild_section)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,193 +0,0 @@
|
||||
"""Structural regression tests for modules/core/local-domain-loopback.nix.
|
||||
|
||||
Verifies that the sovran-hosts-update helper:
|
||||
- is built as a pkgs.writeShellApplication with explicit runtimeInputs
|
||||
(gawk, gnugrep, coreutils);
|
||||
- uses ``lib.getExe hostsUpdateScript`` for both the systemd ExecStart and
|
||||
the activation script so that both contexts share the same Nix-store
|
||||
executable;
|
||||
- does NOT point ExecStart at the raw /etc path;
|
||||
- includes element-calling in the supported domain list;
|
||||
- uses ``awk -v`` for safe marker-variable passing rather than interpolating
|
||||
marker text directly into the awk program;
|
||||
- retains the domain validation regex (idempotency / injection prevention);
|
||||
- exposes /etc/sovran-hosts-update.sh as a symlink via ``source =`` (not a
|
||||
second raw ``text =`` body);
|
||||
- does NOT rely on environment.systemPackages for the helper's dependencies.
|
||||
"""
|
||||
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
NIX_STRING_INDENT = 6
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
FLAKE_SOURCE = (REPO_ROOT / "flake.nix").read_text()
|
||||
PRIMARY_NIXOS_CONFIGURATION_MATCH = re.search(
|
||||
r"nixosConfigurations\.([A-Za-z0-9_-]+)\s*=",
|
||||
FLAKE_SOURCE,
|
||||
)
|
||||
if PRIMARY_NIXOS_CONFIGURATION_MATCH is None:
|
||||
raise RuntimeError("Could not determine the primary nixosConfigurations entry from flake.nix")
|
||||
PRIMARY_NIXOS_CONFIGURATION = PRIMARY_NIXOS_CONFIGURATION_MATCH.group(1)
|
||||
HELPER_BUILD_ATTR = (
|
||||
f'.#nixosConfigurations.{PRIMARY_NIXOS_CONFIGURATION}.config.environment.etc.'
|
||||
'"sovran-hosts-update.sh".source'
|
||||
)
|
||||
NIX_FILE = (
|
||||
REPO_ROOT
|
||||
/ "modules"
|
||||
/ "core"
|
||||
/ "local-domain-loopback.nix"
|
||||
)
|
||||
|
||||
|
||||
class LocalDomainLoopbackNixStructureTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.source = NIX_FILE.read_text()
|
||||
|
||||
def _helper_script(self) -> str:
|
||||
start = self.source.index("text = ''") + len("text = ''")
|
||||
end = self.source.index(" '';", start)
|
||||
return "\n".join(
|
||||
line[NIX_STRING_INDENT:]
|
||||
if line.startswith(" " * NIX_STRING_INDENT)
|
||||
else line
|
||||
for line in self.source[start:end].splitlines()
|
||||
).lstrip("\n")
|
||||
|
||||
# ── writeShellApplication and explicit runtimeInputs ────────────────────
|
||||
|
||||
def test_uses_write_shell_application(self):
|
||||
self.assertIn("pkgs.writeShellApplication", self.source)
|
||||
|
||||
def test_runtime_inputs_includes_coreutils(self):
|
||||
self.assertIn("pkgs.coreutils", self.source)
|
||||
|
||||
def test_runtime_inputs_includes_gawk(self):
|
||||
self.assertIn("pkgs.gawk", self.source)
|
||||
|
||||
def test_runtime_inputs_includes_gnugrep(self):
|
||||
self.assertIn("pkgs.gnugrep", self.source)
|
||||
|
||||
def test_runtime_inputs_block_present(self):
|
||||
self.assertIn("runtimeInputs", self.source)
|
||||
|
||||
# ── Both execution paths use lib.getExe ─────────────────────────────────
|
||||
|
||||
def test_exec_start_uses_lib_get_exe(self):
|
||||
"""systemd ExecStart must reference the Nix-store executable."""
|
||||
self.assertIn("ExecStart = lib.getExe hostsUpdateScript", self.source)
|
||||
|
||||
def test_activation_script_uses_lib_get_exe(self):
|
||||
"""Activation text must call the same Nix-store executable."""
|
||||
self.assertIn("${lib.getExe hostsUpdateScript}", self.source)
|
||||
|
||||
def test_exec_start_does_not_point_to_etc_path(self):
|
||||
"""ExecStart must NOT use the raw /etc path (which lacks a deterministic PATH)."""
|
||||
self.assertNotIn('ExecStart = "/etc/sovran-hosts-update.sh"', self.source)
|
||||
|
||||
# ── /etc symlink uses source =, not a second text = body ────────────────
|
||||
|
||||
def test_etc_entry_uses_source_not_text(self):
|
||||
"""The /etc/sovran-hosts-update.sh entry must be a symlink (source =),
|
||||
not a second raw script body (text =)."""
|
||||
self.assertIn(
|
||||
'environment.etc."sovran-hosts-update.sh".source', self.source
|
||||
)
|
||||
|
||||
def test_etc_source_points_to_get_exe(self):
|
||||
self.assertIn(
|
||||
'environment.etc."sovran-hosts-update.sh".source = lib.getExe hostsUpdateScript',
|
||||
self.source,
|
||||
)
|
||||
|
||||
# ── element-calling domain is supported ─────────────────────────────────
|
||||
|
||||
def test_element_calling_domain_key_present(self):
|
||||
self.assertIn("element-calling", self.source)
|
||||
|
||||
# ── Robust awk -v variable passing ──────────────────────────────────────
|
||||
|
||||
def test_awk_uses_dash_v_for_begin_marker(self):
|
||||
"""awk must receive the begin marker via -v, not by shell interpolation."""
|
||||
self.assertIn('awk -v begin=', self.source)
|
||||
|
||||
def test_awk_uses_dash_v_for_end_marker(self):
|
||||
self.assertIn('-v end=', self.source)
|
||||
|
||||
def test_awk_does_not_interpolate_marker_into_program(self):
|
||||
"""The old pattern interpolated $BEGIN_MARKER directly into the awk source."""
|
||||
self.assertNotIn('/$BEGIN_MARKER', self.source)
|
||||
self.assertNotIn('/$END_MARKER', self.source)
|
||||
|
||||
# ── Domain validation ────────────────────────────────────────────────────
|
||||
|
||||
def test_domain_validation_regex_present(self):
|
||||
"""The hostname validation regex must still be present for injection prevention."""
|
||||
self.assertIn("grep -qE", self.source)
|
||||
self.assertIn("[a-zA-Z0-9]", self.source)
|
||||
|
||||
def test_invalid_domain_warning_present(self):
|
||||
self.assertIn("skipping invalid domain value", self.source)
|
||||
|
||||
# ── No environment.systemPackages reliance ───────────────────────────────
|
||||
|
||||
def test_no_environment_system_packages_for_helper(self):
|
||||
"""The helper's tools are declared via runtimeInputs; the module must
|
||||
not add them to environment.systemPackages."""
|
||||
self.assertNotIn("environment.systemPackages", self.source)
|
||||
|
||||
# ── Idempotency: existing Sovran block is removed before rewriting ───────
|
||||
|
||||
def test_existing_block_removal_logic_present(self):
|
||||
"""awk strip of the managed block must be present for idempotency."""
|
||||
self.assertIn("skip=1", self.source)
|
||||
self.assertIn("skip=0", self.source)
|
||||
|
||||
def test_managed_block_uses_grouped_append_redirect(self):
|
||||
self.assertIn('} >> "$TMP"', self.source)
|
||||
self.assertEqual(self.source.count('>> "$TMP"'), 1)
|
||||
|
||||
def test_helper_script_passes_shellcheck(self):
|
||||
shellcheck = shutil.which("shellcheck")
|
||||
if shellcheck is None:
|
||||
self.skipTest("shellcheck is not installed")
|
||||
proc = subprocess.run(
|
||||
[shellcheck, "-s", "bash", "-"],
|
||||
input=self._helper_script(),
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
output = proc.stdout + proc.stderr
|
||||
self.assertEqual(proc.returncode, 0, output)
|
||||
|
||||
def test_helper_derivation_builds_when_nix_available(self):
|
||||
nix = shutil.which("nix")
|
||||
if nix is None:
|
||||
self.skipTest("nix is not installed")
|
||||
proc = subprocess.run(
|
||||
[
|
||||
nix,
|
||||
"build",
|
||||
HELPER_BUILD_ATTR,
|
||||
"--no-link",
|
||||
],
|
||||
cwd=REPO_ROOT,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
||||
|
||||
# ── Activation script warns on failure rather than silently swallowing ───
|
||||
|
||||
def test_activation_script_emits_warning_on_failure(self):
|
||||
self.assertIn("warning: sovran-hosts-update", self.source)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,399 +0,0 @@
|
||||
"""Tests for server-local loopback diagnostics and domain validation.
|
||||
|
||||
Covers:
|
||||
- Domain value validation and injection prevention.
|
||||
- Loopback address detection (IPv4 and IPv6).
|
||||
- _resolve_all_addresses returning multiple addresses.
|
||||
- _check_domain_health_fast with loopback resolution.
|
||||
- _evaluate_domain_checklist with loopback override — no false dns_mismatch.
|
||||
- _evaluate_domain_checklist with genuine DNS mismatch — still reports error.
|
||||
- api_services health stays "healthy" when domain resolves to loopback.
|
||||
- api_services health stays "needs_attention" when DNS is genuinely wrong.
|
||||
- api_domains_check returns "local_override" for loopback-resolved domains.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, mock_open, patch
|
||||
import sys
|
||||
import types
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Minimal stubs so server.py can be imported without the full FastAPI stack.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _install_web_stubs():
|
||||
if "fastapi" in sys.modules:
|
||||
return
|
||||
|
||||
class _HTTPException(Exception):
|
||||
def __init__(self, status_code=None, detail=None):
|
||||
super().__init__(detail)
|
||||
self.status_code = status_code
|
||||
self.detail = detail
|
||||
|
||||
class _FastAPI:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
def mount(self, *args, **kwargs):
|
||||
return None
|
||||
|
||||
def add_middleware(self, *args, **kwargs):
|
||||
return None
|
||||
|
||||
def __getattr__(self, _name):
|
||||
def _decorator_factory(*args, **kwargs):
|
||||
def _decorator(func):
|
||||
return func
|
||||
return _decorator
|
||||
return _decorator_factory
|
||||
|
||||
class _BaseModel:
|
||||
pass
|
||||
|
||||
class _StaticFiles:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
class _Jinja2Templates:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
class _BaseHTTPMiddleware:
|
||||
pass
|
||||
|
||||
fastapi_module = types.ModuleType("fastapi")
|
||||
fastapi_module.FastAPI = _FastAPI
|
||||
fastapi_module.HTTPException = _HTTPException
|
||||
sys.modules["fastapi"] = fastapi_module
|
||||
|
||||
responses_module = types.ModuleType("fastapi.responses")
|
||||
responses_module.HTMLResponse = object
|
||||
responses_module.JSONResponse = object
|
||||
responses_module.RedirectResponse = object
|
||||
sys.modules["fastapi.responses"] = responses_module
|
||||
|
||||
staticfiles_module = types.ModuleType("fastapi.staticfiles")
|
||||
staticfiles_module.StaticFiles = _StaticFiles
|
||||
sys.modules["fastapi.staticfiles"] = staticfiles_module
|
||||
|
||||
templating_module = types.ModuleType("fastapi.templating")
|
||||
templating_module.Jinja2Templates = _Jinja2Templates
|
||||
sys.modules["fastapi.templating"] = templating_module
|
||||
|
||||
requests_module = types.ModuleType("fastapi.requests")
|
||||
requests_module.Request = object
|
||||
sys.modules["fastapi.requests"] = requests_module
|
||||
|
||||
pydantic_module = types.ModuleType("pydantic")
|
||||
pydantic_module.BaseModel = _BaseModel
|
||||
sys.modules["pydantic"] = pydantic_module
|
||||
|
||||
starlette_base_module = types.ModuleType("starlette.middleware.base")
|
||||
starlette_base_module.BaseHTTPMiddleware = _BaseHTTPMiddleware
|
||||
sys.modules["starlette.middleware.base"] = starlette_base_module
|
||||
|
||||
starlette_middleware_module = types.ModuleType("starlette.middleware")
|
||||
starlette_middleware_module.base = starlette_base_module
|
||||
sys.modules["starlette.middleware"] = starlette_middleware_module
|
||||
|
||||
starlette_module = types.ModuleType("starlette")
|
||||
starlette_module.middleware = starlette_middleware_module
|
||||
sys.modules["starlette"] = starlette_module
|
||||
|
||||
|
||||
_install_web_stubs()
|
||||
from sovran_systemsos_web import server # noqa: E402
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Domain value validation
|
||||
# ===========================================================================
|
||||
|
||||
class TestValidateDomainValue(unittest.TestCase):
|
||||
"""_validate_domain_value must reject anything that could corrupt /etc/hosts."""
|
||||
|
||||
def _v(self, value: str) -> bool:
|
||||
return server._validate_domain_value(value)
|
||||
|
||||
# -- Valid values --------------------------------------------------------
|
||||
|
||||
def test_simple_domain_valid(self):
|
||||
self.assertTrue(self._v("cloud.example.com"))
|
||||
|
||||
def test_subdomain_valid(self):
|
||||
self.assertTrue(self._v("matrix.home.example.org"))
|
||||
|
||||
def test_single_label_with_tld_valid(self):
|
||||
self.assertTrue(self._v("example.com"))
|
||||
|
||||
def test_hyphen_in_domain_valid(self):
|
||||
self.assertTrue(self._v("my-nextcloud.example.com"))
|
||||
|
||||
# -- Injection / malformed values ----------------------------------------
|
||||
|
||||
def test_empty_string_invalid(self):
|
||||
self.assertFalse(self._v(""))
|
||||
|
||||
def test_newline_injection_invalid(self):
|
||||
self.assertFalse(self._v("evil.com\n127.0.0.1 other.host"))
|
||||
|
||||
def test_carriage_return_injection_invalid(self):
|
||||
self.assertFalse(self._v("evil.com\r127.0.0.1 other.host"))
|
||||
|
||||
def test_space_injection_invalid(self):
|
||||
self.assertFalse(self._v("evil.com 127.0.0.1"))
|
||||
|
||||
def test_hash_comment_injection_invalid(self):
|
||||
self.assertFalse(self._v("evil.com# comment"))
|
||||
|
||||
def test_bare_hostname_no_dot_invalid(self):
|
||||
self.assertFalse(self._v("localhost"))
|
||||
|
||||
def test_bare_ip_invalid(self):
|
||||
self.assertFalse(self._v("192.168.1.1"))
|
||||
|
||||
def test_too_long_invalid(self):
|
||||
self.assertFalse(self._v("a" * 254 + ".com"))
|
||||
|
||||
def test_leading_dot_invalid(self):
|
||||
self.assertFalse(self._v(".example.com"))
|
||||
|
||||
def test_trailing_dot_invalid(self):
|
||||
self.assertFalse(self._v("example.com."))
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Loopback address detection
|
||||
# ===========================================================================
|
||||
|
||||
class TestIsLoopbackAddress(unittest.TestCase):
|
||||
|
||||
def test_ipv4_loopback(self):
|
||||
self.assertTrue(server._is_loopback_address("127.0.0.1"))
|
||||
|
||||
def test_ipv4_loopback_other(self):
|
||||
self.assertTrue(server._is_loopback_address("127.0.0.2"))
|
||||
|
||||
def test_ipv4_loopback_high(self):
|
||||
self.assertTrue(server._is_loopback_address("127.255.255.255"))
|
||||
|
||||
def test_ipv6_loopback(self):
|
||||
self.assertTrue(server._is_loopback_address("::1"))
|
||||
|
||||
def test_public_ipv4_not_loopback(self):
|
||||
self.assertFalse(server._is_loopback_address("203.0.113.10"))
|
||||
|
||||
def test_private_ipv4_not_loopback(self):
|
||||
self.assertFalse(server._is_loopback_address("192.168.1.50"))
|
||||
|
||||
def test_ipv6_public_not_loopback(self):
|
||||
self.assertFalse(server._is_loopback_address("2001:db8::1"))
|
||||
|
||||
def test_invalid_string_not_loopback(self):
|
||||
self.assertFalse(server._is_loopback_address("not-an-ip"))
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# _check_domain_health_fast
|
||||
# ===========================================================================
|
||||
|
||||
class TestCheckDomainHealthFast(unittest.TestCase):
|
||||
"""_check_domain_health_fast returns True when there is an issue,
|
||||
False when everything looks fine."""
|
||||
|
||||
def _fast(self, domain, external_ip, resolved_addrs):
|
||||
with patch.object(server, "_resolve_all_addresses", return_value=resolved_addrs):
|
||||
return server._check_domain_health_fast(domain, external_ip)
|
||||
|
||||
def test_no_domain_no_issue(self):
|
||||
# None/empty domain: the fast check reports True (handled by checklist).
|
||||
result = server._check_domain_health_fast(None, "203.0.113.10")
|
||||
self.assertTrue(result)
|
||||
|
||||
def test_empty_domain_no_issue(self):
|
||||
result = server._check_domain_health_fast("", "203.0.113.10")
|
||||
self.assertTrue(result)
|
||||
|
||||
def test_loopback_ipv4_no_issue(self):
|
||||
"""Loopback override must not be flagged as a DNS mismatch."""
|
||||
result = self._fast("cloud.example.com", "203.0.113.10", ["127.0.0.1"])
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_loopback_ipv6_no_issue(self):
|
||||
result = self._fast("cloud.example.com", "203.0.113.10", ["::1"])
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_matches_external_ip_no_issue(self):
|
||||
result = self._fast("cloud.example.com", "203.0.113.10", ["203.0.113.10"])
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_mismatch_is_an_issue(self):
|
||||
result = self._fast("cloud.example.com", "203.0.113.10", ["198.51.100.1"])
|
||||
self.assertTrue(result)
|
||||
|
||||
def test_unavailable_external_ip_no_issue(self):
|
||||
result = self._fast("cloud.example.com", "unavailable", ["198.51.100.1"])
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_multiple_addresses_one_matches_no_issue(self):
|
||||
"""If any resolved address matches external_ip the check should pass."""
|
||||
result = self._fast(
|
||||
"cloud.example.com", "203.0.113.10",
|
||||
["198.51.100.1", "203.0.113.10"],
|
||||
)
|
||||
self.assertFalse(result)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# _evaluate_domain_checklist — loopback override path
|
||||
# ===========================================================================
|
||||
|
||||
class TestEvaluateDomainChecklistLoopback(unittest.TestCase):
|
||||
|
||||
def _eval(self, domain, external_ip, resolved_addrs, reachable_result=None):
|
||||
with (
|
||||
patch.object(server, "_resolve_all_addresses", return_value=resolved_addrs),
|
||||
patch.object(server, "_check_domain_reachable",
|
||||
return_value=reachable_result or {"reachable": True, "status_code": 200}),
|
||||
):
|
||||
return server._evaluate_domain_checklist(domain, external_ip)
|
||||
|
||||
def test_loopback_dns_step_is_ok_not_error(self):
|
||||
result = self._eval("cloud.example.com", "203.0.113.10", ["127.0.0.1"])
|
||||
dns_step = next(s for s in result["domain_check_steps"] if s["step"] == 2)
|
||||
self.assertEqual(dns_step["status"], "ok")
|
||||
self.assertNotIn("mismatch", dns_step.get("detail", "").lower())
|
||||
|
||||
def test_loopback_domain_status_is_local_override(self):
|
||||
result = self._eval("cloud.example.com", "203.0.113.10", ["127.0.0.1"])
|
||||
self.assertEqual(result["domain_status"]["status"], "local_override")
|
||||
|
||||
def test_loopback_has_no_issues_when_reachable(self):
|
||||
result = self._eval(
|
||||
"cloud.example.com", "203.0.113.10", ["127.0.0.1"],
|
||||
reachable_result={"reachable": True, "status_code": 200},
|
||||
)
|
||||
self.assertFalse(result["has_issues"])
|
||||
|
||||
def test_loopback_has_issues_when_caddy_unreachable(self):
|
||||
"""A loopback override with Caddy down should still report an issue."""
|
||||
result = self._eval(
|
||||
"cloud.example.com", "203.0.113.10", ["127.0.0.1"],
|
||||
reachable_result={"reachable": False, "error": "connection refused"},
|
||||
)
|
||||
self.assertTrue(result["has_issues"])
|
||||
|
||||
def test_ipv6_loopback_no_issue(self):
|
||||
result = self._eval("cloud.example.com", "203.0.113.10", ["::1"])
|
||||
self.assertEqual(result["domain_status"]["status"], "local_override")
|
||||
self.assertFalse(result["has_issues"])
|
||||
|
||||
def test_genuine_mismatch_still_reports_error(self):
|
||||
result = self._eval("cloud.example.com", "203.0.113.10", ["198.51.100.1"])
|
||||
self.assertEqual(result["domain_status"]["status"], "dns_mismatch")
|
||||
self.assertTrue(result["has_issues"])
|
||||
|
||||
def test_correct_public_dns_still_reports_ok(self):
|
||||
result = self._eval("cloud.example.com", "203.0.113.10", ["203.0.113.10"])
|
||||
self.assertEqual(result["domain_status"]["status"], "connected")
|
||||
self.assertFalse(result["has_issues"])
|
||||
|
||||
def test_no_domain_has_issues(self):
|
||||
result = self._eval(None, "203.0.113.10", [])
|
||||
self.assertTrue(result["has_issues"])
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# api_services — composite health with loopback
|
||||
# ===========================================================================
|
||||
|
||||
class TestApiServicesLoopbackHealth(unittest.IsolatedAsyncioTestCase):
|
||||
|
||||
async def _get_health(self, resolved_addrs, cached_reachable):
|
||||
"""Return the health value for a single domain-requiring service."""
|
||||
service_cfg = {
|
||||
"services": [
|
||||
{"unit": "caddy.service", "icon": "nextcloud", "enabled": True, "type": "system"}
|
||||
]
|
||||
}
|
||||
with (
|
||||
patch.object(server, "load_config", return_value=service_cfg),
|
||||
patch.object(server, "_read_hub_overrides", return_value=({}, None, None)),
|
||||
patch.object(server.sysctl, "is_active", return_value="active"),
|
||||
patch.dict(server.SERVICE_DOMAIN_MAP, {"caddy.service": "nextcloud"}, clear=False),
|
||||
patch("builtins.open", mock_open(read_data="cloud.example.com\n")),
|
||||
patch.object(server, "_resolve_all_addresses", return_value=resolved_addrs),
|
||||
patch.object(server, "_is_domain_reachable_cached", return_value=cached_reachable),
|
||||
patch.object(server, "_get_listening_ports",
|
||||
return_value={"tcp": {80, 443}, "udp": set()}),
|
||||
patch.object(server, "_get_firewall_allowed_ports",
|
||||
return_value={"tcp": set(), "udp": set()}),
|
||||
patch.object(server, "_cached_external_ip", "203.0.113.10"),
|
||||
):
|
||||
results = await server.api_services()
|
||||
|
||||
return results[0]["health"]
|
||||
|
||||
async def test_loopback_and_reachable_is_healthy(self):
|
||||
"""Loopback override + Caddy reachable → healthy, not needs_attention."""
|
||||
health = await self._get_health(["127.0.0.1"], cached_reachable=True)
|
||||
self.assertEqual(health, "healthy")
|
||||
|
||||
async def test_loopback_and_caddy_down_is_needs_attention(self):
|
||||
"""Loopback override + Caddy unreachable → needs_attention (genuine issue)."""
|
||||
health = await self._get_health(["127.0.0.1"], cached_reachable=False)
|
||||
self.assertEqual(health, "needs_attention")
|
||||
|
||||
async def test_correct_dns_and_reachable_is_healthy(self):
|
||||
health = await self._get_health(["203.0.113.10"], cached_reachable=True)
|
||||
self.assertEqual(health, "healthy")
|
||||
|
||||
async def test_dns_mismatch_is_needs_attention(self):
|
||||
health = await self._get_health(["198.51.100.1"], cached_reachable=True)
|
||||
self.assertEqual(health, "needs_attention")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# api_domains_check — loopback detection
|
||||
# ===========================================================================
|
||||
|
||||
class TestApiDomainsCheckLoopback(unittest.IsolatedAsyncioTestCase):
|
||||
|
||||
async def _check(self, resolved_addrs, external_ip="203.0.113.10"):
|
||||
with (
|
||||
patch.object(server, "_resolve_all_addresses", return_value=resolved_addrs),
|
||||
patch.object(server, "_cached_external_ip", external_ip),
|
||||
):
|
||||
result = await server.api_domains_check(
|
||||
MagicMock(domains=["cloud.example.com"])
|
||||
)
|
||||
return result["domains"][0]
|
||||
|
||||
async def test_loopback_ipv4_returns_local_override(self):
|
||||
result = await self._check(["127.0.0.1"])
|
||||
self.assertEqual(result["status"], "local_override")
|
||||
|
||||
async def test_loopback_ipv6_returns_local_override(self):
|
||||
result = await self._check(["::1"])
|
||||
self.assertEqual(result["status"], "local_override")
|
||||
|
||||
async def test_correct_dns_returns_connected(self):
|
||||
result = await self._check(["203.0.113.10"])
|
||||
self.assertEqual(result["status"], "connected")
|
||||
|
||||
async def test_mismatch_returns_dns_mismatch(self):
|
||||
result = await self._check(["198.51.100.1"])
|
||||
self.assertEqual(result["status"], "dns_mismatch")
|
||||
|
||||
async def test_no_resolution_returns_unresolvable(self):
|
||||
result = await self._check([])
|
||||
self.assertEqual(result["status"], "unresolvable")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,122 +0,0 @@
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
RDP_NIX = Path(__file__).resolve().parents[2] / "modules" / "rdp.nix"
|
||||
USERNAME_READ = "USERNAME=\"$(tr -d '\\n' < \"$USERNAME_FILE\")\""
|
||||
USERNAME_LENGTH_GUARD = "if [ \"''${#USERNAME}\" -gt 32 ]; then"
|
||||
SHORT_PASSWORD_GUARD = 'if [ "\'\'${#PASSWORD}" -lt 8 ]; then'
|
||||
|
||||
|
||||
def _section(source: str, start: str, end: str) -> str:
|
||||
start_idx = source.find(start)
|
||||
if start_idx == -1:
|
||||
raise AssertionError(f"Expected section start not found: {start!r}")
|
||||
end_idx = source.find(end, start_idx)
|
||||
if end_idx == -1:
|
||||
raise AssertionError(f"Expected section end not found: {end!r}")
|
||||
return source[start_idx:end_idx]
|
||||
|
||||
|
||||
class RdpModuleBootSetupTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.source = RDP_NIX.read_text()
|
||||
self.gnome_service = _section(
|
||||
self.source,
|
||||
"systemd.services.gnome-remote-desktop = {",
|
||||
"systemd.tmpfiles.rules = [",
|
||||
)
|
||||
self.setup_service = _section(
|
||||
self.source,
|
||||
"systemd.services.gnome-remote-desktop-setup = {",
|
||||
"};\n}",
|
||||
)
|
||||
|
||||
def test_does_not_redeclare_gnome_remote_desktop_user(self):
|
||||
self.assertNotIn("users.users.gnome-remote-desktop", self.source)
|
||||
self.assertNotIn("createHome = true;", self.source)
|
||||
|
||||
def test_main_service_requires_setup_before_starting(self):
|
||||
self.assertIn('wantedBy = [ "graphical.target" ];', self.gnome_service)
|
||||
self.assertIn('after = [ "gnome-remote-desktop-setup.service" ];', self.gnome_service)
|
||||
self.assertIn('requires = [ "gnome-remote-desktop-setup.service" ];', self.gnome_service)
|
||||
|
||||
def test_setup_waits_for_configuration_service_and_bounded_timeout(self):
|
||||
self.assertIn('wantedBy = [ "graphical.target" ];', self.setup_service)
|
||||
self.assertIn('before = [ "gnome-remote-desktop.service" ];', self.setup_service)
|
||||
self.assertIn('"dbus.service"', self.setup_service)
|
||||
self.assertIn('"gnome-remote-desktop-configuration.service"', self.setup_service)
|
||||
self.assertNotIn("RemainAfterExit", self.setup_service)
|
||||
self.assertIn('TimeoutStartSec = "2min";', self.setup_service)
|
||||
self.assertIn('timeout --kill-after=5s 10s', self.setup_service)
|
||||
self.assertIn('echo "grdctl command timed out: $*" >&2', self.setup_service)
|
||||
self.assertIn('echo "grdctl command failed (exit $rc): $*" >&2', self.setup_service)
|
||||
|
||||
def test_setup_runs_grdctl_directly_as_root(self):
|
||||
# The oneshot service runs as root; grdctl --system is called directly.
|
||||
# GRD 50.x invokes pkexec internally, but the call itself is plain
|
||||
# grdctl --system, not a manual pkexec invocation.
|
||||
self.assertIn('grdctl --system "$@"', self.setup_service)
|
||||
self.assertNotIn("runuser", self.setup_service)
|
||||
self.assertNotIn("sudo", self.setup_service)
|
||||
# No direct Nix-store pkexec invocation (pkgs.polkit}/bin/pkexec).
|
||||
self.assertNotIn("pkgs.polkit}/bin/pkexec", self.setup_service)
|
||||
|
||||
def test_privilege_escalation_packages_absent_from_setup_path(self):
|
||||
self.assertNotIn("pkgs.polkit", self.setup_service)
|
||||
self.assertNotIn("pkgs.util-linux", self.setup_service)
|
||||
|
||||
def test_run_wrappers_bin_prepended_to_path(self):
|
||||
# /run/wrappers/bin must be prepended to PATH before any grdctl_system
|
||||
# invocation so that grdctl --system resolves the NixOS setuid pkexec.
|
||||
path_export = 'export PATH="/run/wrappers/bin:$PATH"'
|
||||
grdctl_marker = "grdctl_system"
|
||||
script = self.setup_service
|
||||
path_idx = script.find(path_export)
|
||||
grdctl_idx = script.find(grdctl_marker)
|
||||
self.assertGreater(path_idx, -1, f"{path_export!r} not found in setup script")
|
||||
self.assertGreater(
|
||||
grdctl_idx, path_idx,
|
||||
"PATH export must appear before the first grdctl_system usage",
|
||||
)
|
||||
|
||||
def test_pkexec_preflight_check(self):
|
||||
# A preflight must confirm /run/wrappers/bin/pkexec is executable
|
||||
# with a clear error message before any GRD configuration changes.
|
||||
self.assertIn("test -x /run/wrappers/bin/pkexec", self.setup_service)
|
||||
self.assertIn(
|
||||
"/run/wrappers/bin/pkexec is absent or not executable",
|
||||
self.setup_service,
|
||||
)
|
||||
|
||||
def test_hub_files_are_the_source_of_truth_for_username_and_password(self):
|
||||
self.assertIn('DEFAULT_USERNAME="sovran"', self.setup_service)
|
||||
self.assertIn('if [ ! -f "$USERNAME_FILE" ]; then', self.setup_service)
|
||||
self.assertIn(USERNAME_READ, self.setup_service)
|
||||
self.assertIn(USERNAME_LENGTH_GUARD, self.setup_service)
|
||||
self.assertIn('case "$USERNAME" in', self.setup_service)
|
||||
self.assertIn('[A-Za-z_][A-Za-z0-9_-]*)', self.setup_service)
|
||||
self.assertIn("RDP username is too long (''${#USERNAME} characters, maximum 32)", self.setup_service)
|
||||
self.assertIn("RDP username must start with a letter or underscore and contain only letters, numbers, underscores, and hyphens", self.setup_service)
|
||||
self.assertIn('if [ ! -f "$PASSWORD_FILE" ]; then', self.setup_service)
|
||||
self.assertIn("tr -d '\\n'", self.setup_service)
|
||||
self.assertIn('"$PASSWORD_FILE"', self.setup_service)
|
||||
self.assertIn(SHORT_PASSWORD_GUARD, self.setup_service)
|
||||
self.assertIn("RDP password is too short (''${#PASSWORD} characters, minimum 8)", self.setup_service)
|
||||
self.assertIn('grdctl_system rdp set-credentials "$USERNAME" "$PASSWORD"', self.setup_service)
|
||||
self.assertNotIn('grdctl --system rdp set-credentials sovran "$PASSWORD"', self.setup_service)
|
||||
|
||||
def test_secure_permissions_are_enforced_for_state_and_secret_files(self):
|
||||
self.assertIn('"d /var/lib/gnome-remote-desktop/tls 0700', self.source)
|
||||
self.assertIn("chmod 700", self.setup_service)
|
||||
self.assertIn('chmod 600 "$USERNAME_FILE"', self.setup_service)
|
||||
self.assertIn('chmod 600 "$PASSWORD_FILE"', self.setup_service)
|
||||
self.assertIn('chmod 600 "$CRED_FILE"', self.setup_service)
|
||||
self.assertIn('chmod 600 "$TLS_DIR/rdp-tls.key"', self.setup_service)
|
||||
self.assertIn('chmod 644 "$TLS_DIR/rdp-tls.crt"', self.setup_service)
|
||||
self.assertIn('LOCAL_IP="$(hostname -I | awk \'{print $1}\')"', self.setup_service)
|
||||
self.assertIn('LOCAL_IP="127.0.0.1"', self.setup_service)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,171 +0,0 @@
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import mock_open, patch
|
||||
import sys
|
||||
import types
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
|
||||
def _install_web_stubs():
|
||||
if "fastapi" in sys.modules:
|
||||
return
|
||||
|
||||
class _HTTPException(Exception):
|
||||
def __init__(self, status_code=None, detail=None):
|
||||
super().__init__(detail)
|
||||
self.status_code = status_code
|
||||
self.detail = detail
|
||||
|
||||
class _FastAPI:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
def mount(self, *args, **kwargs):
|
||||
return None
|
||||
|
||||
def add_middleware(self, *args, **kwargs):
|
||||
return None
|
||||
|
||||
def __getattr__(self, _name):
|
||||
def _decorator_factory(*args, **kwargs):
|
||||
def _decorator(func):
|
||||
return func
|
||||
|
||||
return _decorator
|
||||
|
||||
return _decorator_factory
|
||||
|
||||
class _BaseModel:
|
||||
pass
|
||||
|
||||
class _StaticFiles:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
class _Jinja2Templates:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
class _BaseHTTPMiddleware:
|
||||
pass
|
||||
|
||||
fastapi_module = types.ModuleType("fastapi")
|
||||
fastapi_module.FastAPI = _FastAPI
|
||||
fastapi_module.HTTPException = _HTTPException
|
||||
sys.modules["fastapi"] = fastapi_module
|
||||
|
||||
responses_module = types.ModuleType("fastapi.responses")
|
||||
responses_module.HTMLResponse = object
|
||||
responses_module.JSONResponse = object
|
||||
responses_module.RedirectResponse = object
|
||||
sys.modules["fastapi.responses"] = responses_module
|
||||
|
||||
staticfiles_module = types.ModuleType("fastapi.staticfiles")
|
||||
staticfiles_module.StaticFiles = _StaticFiles
|
||||
sys.modules["fastapi.staticfiles"] = staticfiles_module
|
||||
|
||||
templating_module = types.ModuleType("fastapi.templating")
|
||||
templating_module.Jinja2Templates = _Jinja2Templates
|
||||
sys.modules["fastapi.templating"] = templating_module
|
||||
|
||||
requests_module = types.ModuleType("fastapi.requests")
|
||||
requests_module.Request = object
|
||||
sys.modules["fastapi.requests"] = requests_module
|
||||
|
||||
pydantic_module = types.ModuleType("pydantic")
|
||||
pydantic_module.BaseModel = _BaseModel
|
||||
sys.modules["pydantic"] = pydantic_module
|
||||
|
||||
starlette_base_module = types.ModuleType("starlette.middleware.base")
|
||||
starlette_base_module.BaseHTTPMiddleware = _BaseHTTPMiddleware
|
||||
sys.modules["starlette.middleware.base"] = starlette_base_module
|
||||
|
||||
starlette_middleware_module = types.ModuleType("starlette.middleware")
|
||||
starlette_middleware_module.base = starlette_base_module
|
||||
sys.modules["starlette.middleware"] = starlette_middleware_module
|
||||
|
||||
starlette_module = types.ModuleType("starlette")
|
||||
starlette_module.middleware = starlette_middleware_module
|
||||
sys.modules["starlette"] = starlette_module
|
||||
|
||||
|
||||
_install_web_stubs()
|
||||
from sovran_systemsos_web import server
|
||||
|
||||
|
||||
class ServiceDetailRouterWordingTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_livekit_service_detail_includes_internal_ip(self):
|
||||
service_cfg = {
|
||||
"services": [
|
||||
{"unit": "livekit.service", "icon": "element-call", "enabled": True, "type": "system"}
|
||||
]
|
||||
}
|
||||
domain_eval = {
|
||||
"domain_status": {"status": "ok"},
|
||||
"domain_reachable": {"reachable": True},
|
||||
"domain_check_steps": [],
|
||||
"has_issues": False,
|
||||
}
|
||||
|
||||
with (
|
||||
patch.object(server, "load_config", return_value=service_cfg),
|
||||
patch.object(server, "_read_hub_overrides", return_value=({}, None, None)),
|
||||
patch.object(server.sysctl, "is_active", return_value="active"),
|
||||
patch.dict(server.SERVICE_DOMAIN_MAP, {"livekit.service": "element-call"}, clear=False),
|
||||
patch.dict(
|
||||
server.SERVICE_PORT_REQUIREMENTS,
|
||||
{"livekit.service": [{"port": "7881", "protocol": "TCP", "description": "LiveKit"}]},
|
||||
clear=False,
|
||||
),
|
||||
patch("builtins.open", mock_open(read_data="call.example.com\n")),
|
||||
patch.object(server, "_evaluate_domain_checklist", return_value=domain_eval),
|
||||
patch.object(server, "_get_internal_ip", return_value="192.168.1.44"),
|
||||
patch.object(server, "_save_internal_ip"),
|
||||
patch.object(server, "_get_listening_ports", return_value={"tcp": {7881}, "udp": set()}),
|
||||
patch.object(server, "_get_firewall_allowed_ports", return_value={"tcp": set(), "udp": set()}),
|
||||
):
|
||||
result = await server.api_service_detail("livekit.service")
|
||||
|
||||
self.assertEqual(result["internal_ip"], "192.168.1.44")
|
||||
self.assertEqual(result["extra_ports"][0]["status"], "listening")
|
||||
self.assertEqual(result["domain_check_steps"][-1]["label"], "Router Setup Needed")
|
||||
|
||||
async def test_livekit_router_step_uses_not_ready_yet_wording(self):
|
||||
service_cfg = {
|
||||
"services": [
|
||||
{"unit": "livekit.service", "icon": "element-call", "enabled": True, "type": "system"}
|
||||
]
|
||||
}
|
||||
domain_eval = {
|
||||
"domain_status": {"status": "ok"},
|
||||
"domain_reachable": {"reachable": True},
|
||||
"domain_check_steps": [],
|
||||
"has_issues": False,
|
||||
}
|
||||
|
||||
with (
|
||||
patch.object(server, "load_config", return_value=service_cfg),
|
||||
patch.object(server, "_read_hub_overrides", return_value=({}, None, None)),
|
||||
patch.object(server.sysctl, "is_active", return_value="active"),
|
||||
patch.dict(server.SERVICE_DOMAIN_MAP, {"livekit.service": "element-call"}, clear=False),
|
||||
patch.dict(
|
||||
server.SERVICE_PORT_REQUIREMENTS,
|
||||
{"livekit.service": [{"port": "7881", "protocol": "TCP", "description": "LiveKit"}]},
|
||||
clear=False,
|
||||
),
|
||||
patch("builtins.open", mock_open(read_data="call.example.com\n")),
|
||||
patch.object(server, "_evaluate_domain_checklist", return_value=domain_eval),
|
||||
patch.object(server, "_get_internal_ip", return_value="192.168.1.44"),
|
||||
patch.object(server, "_save_internal_ip"),
|
||||
patch.object(server, "_get_listening_ports", return_value={"tcp": set(), "udp": set()}),
|
||||
patch.object(server, "_get_firewall_allowed_ports", return_value={"tcp": set(), "udp": set()}),
|
||||
):
|
||||
result = await server.api_service_detail("livekit.service")
|
||||
|
||||
self.assertEqual(result["extra_ports"][0]["status"], "closed")
|
||||
self.assertIn("Not ready yet", result["domain_check_steps"][-1]["detail"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,99 +0,0 @@
|
||||
"""Regression test for Starlette 1.1.0+ TemplateResponse keyword-argument style.
|
||||
|
||||
Prior to this fix, the three HTML routes called:
|
||||
templates.TemplateResponse("name.html", {"request": request, ...})
|
||||
which passes the context dict as the second positional argument. With the
|
||||
updated Starlette/FastAPI versions shipped in NixOS unstable (Starlette 1.1.0,
|
||||
FastAPI 0.136.3) that positional argument is the template name, causing Jinja2
|
||||
to receive a dict as a cache key and raise:
|
||||
TypeError: unhashable type: 'dict'
|
||||
|
||||
The fix updates every call to use keyword arguments:
|
||||
templates.TemplateResponse(request=request, name="name.html", context={...})
|
||||
"""
|
||||
|
||||
import ast
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
SERVER_PY = Path(__file__).resolve().parents[1] / "sovran_systemsos_web" / "server.py"
|
||||
|
||||
|
||||
def _template_response_calls(source: str):
|
||||
"""Return a list of ast.Call nodes that are TemplateResponse calls."""
|
||||
tree = ast.parse(source)
|
||||
calls = []
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.Call):
|
||||
continue
|
||||
func = node.func
|
||||
if isinstance(func, ast.Attribute) and func.attr == "TemplateResponse":
|
||||
calls.append(node)
|
||||
return calls
|
||||
|
||||
|
||||
class TemplateResponseSignatureTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.source = SERVER_PY.read_text()
|
||||
self.calls = _template_response_calls(self.source)
|
||||
|
||||
def test_at_least_one_template_response_call_found(self):
|
||||
self.assertGreater(len(self.calls), 0, "No TemplateResponse calls found in server.py")
|
||||
|
||||
def test_no_old_style_positional_dict_context(self):
|
||||
"""No TemplateResponse call should pass a dict literal as its second positional arg.
|
||||
|
||||
The old style was:
|
||||
templates.TemplateResponse("name.html", {"request": request, ...})
|
||||
where args[0] is a string and args[1] is a Dict node. That pattern
|
||||
triggers the Starlette 1.1.0 bug.
|
||||
"""
|
||||
for call in self.calls:
|
||||
positional = call.args
|
||||
if len(positional) >= 2 and isinstance(positional[1], ast.Dict):
|
||||
self.fail(
|
||||
f"Found old-style TemplateResponse call at line {call.lineno}: "
|
||||
"second positional argument is a dict literal. "
|
||||
"Use keyword arguments (request=, name=, context=) instead."
|
||||
)
|
||||
|
||||
def test_request_not_duplicated_in_context(self):
|
||||
"""The 'request' key must not appear inside the context= dict when
|
||||
request= is already passed as a dedicated keyword argument."""
|
||||
for call in self.calls:
|
||||
kw_dict = {kw.arg: kw.value for kw in call.keywords if isinstance(kw, ast.keyword)}
|
||||
|
||||
if "request" not in kw_dict:
|
||||
continue # no request= kwarg, nothing to check
|
||||
|
||||
context_node = kw_dict.get("context")
|
||||
if not isinstance(context_node, ast.Dict):
|
||||
continue
|
||||
|
||||
for key_node in context_node.keys:
|
||||
if isinstance(key_node, ast.Constant) and key_node.value == "request":
|
||||
self.fail(
|
||||
f"TemplateResponse at line {call.lineno} passes 'request' both as "
|
||||
"request= keyword argument and inside the context dict."
|
||||
)
|
||||
|
||||
def test_all_calls_use_keyword_arguments(self):
|
||||
"""Every TemplateResponse call should use keyword arguments for request, name,
|
||||
and context rather than relying on positional ordering."""
|
||||
for call in self.calls:
|
||||
kw_args = {kw.arg for kw in call.keywords if isinstance(kw, ast.keyword)}
|
||||
self.assertIn(
|
||||
"request",
|
||||
kw_args,
|
||||
f"TemplateResponse at line {call.lineno} is missing keyword argument 'request='.",
|
||||
)
|
||||
self.assertIn(
|
||||
"name",
|
||||
kw_args,
|
||||
f"TemplateResponse at line {call.lineno} is missing keyword argument 'name='.",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user