gitea-admin/gh-christianlouis-dmarq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

address copilot review followups (#96)

Browse Source
This commit is contained in:
Christian Krakau-Louis
2026-05-18 19:17:36 +02:00
committed by GitHub
parent ee7c15ce4b
commit 181b43bff9
12 changed files with 245 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/api_v1/endpoints/domains.py
+16 -6
View File
@@ -20,6 +20,7 @@ from app.services.report_store import ReportStore
logger = logging.getLogger(__name__)
router = APIRouter()
DOMAIN_SELECTOR_LOOKUP_CHUNK_SIZE = 500
class DomainBase(BaseModel):
@@ -145,17 +146,26 @@ def _get_domain_selectors_from_db(db: Session, domain_name: str) -> List[str]:
return []
def _get_domain_selectors_map_from_db(db: Session, domain_names: List[str]) -> Dict[str, List[str]]:
def _get_domain_selectors_map_from_db(
db: Session, domain_names: List[str]
) -> Dict[str, List[str]]:
"""Return manually configured DKIM selectors for all requested domains."""
if not domain_names:
return {}
rows = db.query(Domain.name, Domain.dkim_selectors).filter(Domain.name.in_(domain_names)).all()
unique_names = list(dict.fromkeys(domain_names))
selectors_by_domain: Dict[str, List[str]] = {}
for name, selectors in rows:
selectors_by_domain[name] = [
selector.strip() for selector in (selectors or "").split(",") if selector.strip()
]
for index in range(0, len(unique_names), DOMAIN_SELECTOR_LOOKUP_CHUNK_SIZE):
chunk = unique_names[index : index + DOMAIN_SELECTOR_LOOKUP_CHUNK_SIZE]
rows = (
db.query(Domain.name, Domain.dkim_selectors)
.filter(Domain.name.in_(chunk))
.all()
)
for name, selectors in rows:
selectors_by_domain[name] = [
selector.strip() for selector in (selectors or "").split(",") if selector.strip()
]
return selectors_by_domain
backend/app/api/api_v1/endpoints/setup.py
+106 -5
View File
@@ -3,8 +3,11 @@ from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Request, Security, status
from fastapi.security import HTTPAuthorizationCredentials
from pydantic import BaseModel, EmailStr
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.security import api_key_header, require_admin_auth, security_bearer
from app.models.setting import Setting
router = APIRouter()
@@ -15,6 +18,71 @@ setup_status = {
"app_name": "DMARQ",
}
SETUP_COMPLETE_KEY = "setup.is_complete"
SETUP_ADMIN_EMAIL_KEY = "setup.admin_email"
GENERAL_APP_NAME_KEY = "general.app_name"
GENERAL_BASE_URL_KEY = "general.base_url"
def _setting_value(db: Session, key: str) -> Optional[str]:
row = db.query(Setting).filter(Setting.key == key).first()
return row.value if row else None
def _is_true(value: Optional[str]) -> bool:
return str(value).strip().lower() in {"1", "true", "yes", "on"}
def _upsert_setting(
db: Session,
key: str,
value: Optional[str],
*,
description: str,
value_type: str = "string",
category: str = "setup",
) -> None:
row = db.query(Setting).filter(Setting.key == key).first()
if row is None:
db.add(
Setting(
key=key,
value=value,
description=description,
value_type=value_type,
category=category,
)
)
return
row.value = value
row.description = row.description or description
row.value_type = row.value_type or value_type
row.category = row.category or category
def _refresh_setup_status_from_db(db: Session) -> dict:
"""Merge persisted setup state into the legacy in-memory setup status."""
persisted_complete = _is_true(_setting_value(db, SETUP_COMPLETE_KEY))
if persisted_complete:
setup_status["is_setup_complete"] = True
persisted_admin_email = _setting_value(db, SETUP_ADMIN_EMAIL_KEY)
if persisted_admin_email:
setup_status["admin_email"] = persisted_admin_email
persisted_app_name = _setting_value(db, GENERAL_APP_NAME_KEY)
if persisted_app_name:
setup_status["app_name"] = persisted_app_name
return setup_status
def _setup_is_complete(db: Session) -> bool:
return bool(setup_status["is_setup_complete"]) or _is_true(
_setting_value(db, SETUP_COMPLETE_KEY)
)
class SetupStatusResponse(BaseModel):
"""Setup status response"""
@@ -40,40 +108,50 @@ class SystemConfigRequest(BaseModel):
async def require_setup_write_auth(
request: Request,
db: Session = Depends(get_db),
api_key: Optional[str] = Security(api_key_header),
bearer: Optional[HTTPAuthorizationCredentials] = Security(security_bearer),
) -> dict:
"""Allow unauthenticated first-time setup writes, then require admin auth."""
if not setup_status["is_setup_complete"]:
if not _setup_is_complete(db):
return {"auth_type": "initial_setup"}
return await require_admin_auth(request=request, api_key=api_key, bearer=bearer)
@router.get("/status", response_model=SetupStatusResponse)
async def get_setup_status():
async def get_setup_status(db: Session = Depends(get_db)):
"""Get the current setup status"""
current_status = _refresh_setup_status_from_db(db)
return SetupStatusResponse(
is_setup_complete=setup_status["is_setup_complete"],
app_name=setup_status["app_name"],
is_setup_complete=current_status["is_setup_complete"],
app_name=current_status["app_name"],
)
@router.post("/admin", status_code=201)
async def setup_admin(
request: AdminSetupRequest,
db: Session = Depends(get_db),
_auth: dict = Depends(require_setup_write_auth),
):
"""
Setup admin user during initial system configuration.
For Milestone 1, this simply stores the admin email in memory.
"""
if setup_status["is_setup_complete"]:
if _setup_is_complete(db):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="Setup already completed"
)
# Store admin email
setup_status["admin_email"] = request.email
_upsert_setting(
db,
SETUP_ADMIN_EMAIL_KEY,
request.email,
description="Email address configured during initial setup",
)
db.commit()
return {"message": "Admin user setup completed"}
@@ -81,6 +159,7 @@ async def setup_admin(
@router.post("/system", status_code=200)
async def setup_system(
request: SystemConfigRequest,
db: Session = Depends(get_db),
_auth: dict = Depends(require_setup_write_auth),
):
"""
@@ -90,5 +169,27 @@ async def setup_system(
# Store app name
setup_status["app_name"] = request.app_name
setup_status["is_setup_complete"] = True
_upsert_setting(
db,
GENERAL_APP_NAME_KEY,
request.app_name,
description="Application display name shown in the UI",
category="general",
)
_upsert_setting(
db,
GENERAL_BASE_URL_KEY,
request.base_url,
description="Public base URL for this DMARQ instance",
category="general",
)
_upsert_setting(
db,
SETUP_COMPLETE_KEY,
"true",
description="Whether initial setup has been completed",
value_type="boolean",
)
db.commit()
return {"message": "System settings saved successfully"}
backend/app/tests/test_config.py
+2 -1
View File
@@ -173,7 +173,8 @@ class TestAdminApiKeySetting:
class TestLogtoSettings:
def test_ssl_verification_is_enabled_by_default(self):
def test_ssl_verification_is_enabled_by_default(self, monkeypatch):
monkeypatch.delenv("LOGTO_SKIP_SSL_VERIFY", raising=False)
settings = Settings()
assert settings.LOGTO_SKIP_SSL_VERIFY is False
backend/app/tests/test_dns_endpoints.py
+46
View File
@@ -13,6 +13,8 @@ from unittest.mock import AsyncMock, patch
import pytest
from fastapi.testclient import TestClient
from app.api.api_v1.endpoints import domains as domains_endpoint
from app.models.domain import Domain
from app.services.dns_resolver import DomainDNSResult
from app.services.report_store import ReportStore
@@ -290,6 +292,50 @@ def test_summary_dns_failure_defaults_false(client: TestClient):
assert domain["dkim_status"] is False
def test_summary_endpoint_uses_manual_selectors(client: TestClient):
"""Manually configured selectors are forwarded by the summary endpoint."""
client.post(f"/api/v1/domains/{DOMAIN}/selectors", json={"selector": "manualsel"})
captured_selectors = []
async def _fake_check_domain(domain, selectors=None):
captured_selectors.extend(selectors or [])
return MOCK_DNS_RESULT
with patch(
"app.api.api_v1.endpoints.domains.get_default_provider",
return_value=AsyncMock(check_domain=_fake_check_domain),
):
response = client.get("/api/v1/domains/summary")
assert response.status_code == 200
assert "manualsel" in captured_selectors
assert "google" in captured_selectors
def test_selector_map_lookup_chunks_domain_names(db_session, monkeypatch):
"""Large summary batches are split to avoid database parameter limits."""
monkeypatch.setattr(domains_endpoint, "DOMAIN_SELECTOR_LOOKUP_CHUNK_SIZE", 2)
db_session.add_all(
[
Domain(name="one.example", dkim_selectors="a,b"),
Domain(name="two.example", dkim_selectors="c"),
Domain(name="three.example", dkim_selectors="d"),
]
)
db_session.commit()
selectors = domains_endpoint._get_domain_selectors_map_from_db(
db_session,
["one.example", "two.example", "three.example", "one.example"],
)
assert selectors == {
"one.example": ["a", "b"],
"two.example": ["c"],
"three.example": ["d"],
}
# ---------------------------------------------------------------------------
# GET /api/v1/domains/{domain_id}/sources (PTR + fix hints)
# ---------------------------------------------------------------------------
backend/app/tests/test_main_polling.py
+2 -2
View File
@@ -21,9 +21,9 @@ class TestNextSleepSeconds:
def test_respects_min_sleep(self):
from app.main import _next_sleep_seconds
source = SimpleNamespace(polling_interval=0)
source = SimpleNamespace(polling_interval=1)
assert _next_sleep_seconds(min_sleep=120, enabled_sources=[source]) == 3600
assert _next_sleep_seconds(min_sleep=120, enabled_sources=[source]) == 120
def test_queries_database_when_sources_not_supplied(self):
from app.main import _next_sleep_seconds
backend/app/tests/test_setup_endpoints.py
+50
View File
@@ -9,6 +9,7 @@ from fastapi.testclient import TestClient
from app.api.api_v1.endpoints.setup import setup_status
from app.core.security import _api_keys, add_api_key
from app.models.setting import Setting
@pytest.fixture(autouse=True)
@@ -42,6 +43,38 @@ class TestSetupStatus:
assert data["is_setup_complete"] is True
assert data["app_name"] == "MyApp"
def test_status_reads_persisted_setup_state_after_memory_reset(
self, client: TestClient, db_session
):
db_session.add(
Setting(
key="setup.is_complete",
value="true",
description="Whether initial setup has been completed",
value_type="boolean",
category="setup",
)
)
db_session.add(
Setting(
key="general.app_name",
value="Persisted DMARQ",
description="Application display name shown in the UI",
value_type="string",
category="general",
)
)
db_session.commit()
setup_status["is_setup_complete"] = False
setup_status["app_name"] = "DMARQ"
response = client.get("/api/v1/setup/status")
assert response.status_code == 200
data = response.json()
assert data["is_setup_complete"] is True
assert data["app_name"] == "Persisted DMARQ"
class TestSetupAdmin:
"""Tests for POST /api/v1/setup/admin"""
@@ -137,6 +170,23 @@ class TestSetupSystem:
assert response.status_code == 401
def test_system_setup_requires_auth_if_persisted_complete_after_restart(
self, client: TestClient
):
first_response = client.post(
"/api/v1/setup/system",
json={"app_name": "DMARQ", "base_url": "https://example.com"},
)
assert first_response.status_code == 200
setup_status["is_setup_complete"] = False
response = client.post(
"/api/v1/setup/system",
json={"app_name": "Changed", "base_url": "https://changed.example.com"},
)
assert response.status_code == 401
def test_system_setup_allows_authenticated_updates_after_complete(self, client: TestClient):
setup_status["is_setup_complete"] = True
api_key = "b" * 64