fix: resolve all ruff lint errors and SQLite UNIQUE constraint issue

Ruff fixes (20 errors → 0):
- F401: Remove unused imports (DeckyConfig, random_hostname, IniConfig,
  COMPOSE_FILE, sys, patch) across cli.py, mutator/engine.py,
  templates/ftp, templates/rdp, test_mysql.py, test_postgres.py
- F541: Remove extraneous f-prefixes on strings with no placeholders
  in templates/imap, test_ftp_live, test_http_live
- E741: Rename ambiguous variable 'l' to descriptive names (line, entry,
  part) across conftest.py, test_ftp_live, test_http_live,
  test_mongodb_live, test_pop3, test_ssh

SQLite fix:
- Change _initialize_sync() admin seeding from SELECT-then-INSERT to
  INSERT OR IGNORE, preventing IntegrityError when admin user already
  exists from a previous run
This commit is contained in:
2026-04-12 02:17:50 -04:00
parent 99be4e64ad
commit f78104e1c8
14 changed files with 29 additions and 38 deletions

View File

@@ -24,13 +24,11 @@ from decnet.env import (
)
from decnet.archetypes import Archetype, all_archetypes, get_archetype
from decnet.config import (
DeckyConfig,
DecnetConfig,
random_hostname,
)
from decnet.distros import all_distros, get_distro
from decnet.fleet import all_service_names, build_deckies, build_deckies_from_ini
from decnet.ini_loader import IniConfig, load_ini
from decnet.ini_loader import load_ini
from decnet.network import detect_interface, detect_subnet, allocate_ips, get_host_ip
from decnet.services.registry import all_services

View File

@@ -13,7 +13,7 @@ from decnet.archetypes import get_archetype
from decnet.fleet import all_service_names
from decnet.composer import write_compose
from decnet.config import DeckyConfig, load_state, save_state
from decnet.engine import COMPOSE_FILE, _compose_with_retry
from decnet.engine import _compose_with_retry
import subprocess # nosec B404

View File

@@ -33,14 +33,9 @@ class SQLiteRepository(BaseRepository):
from decnet.web.db.sqlite.database import get_sync_engine
engine = get_sync_engine(self.db_path)
with engine.connect() as conn:
result = conn.execute(
text("SELECT uuid FROM users WHERE username = :u"),
{"u": DECNET_ADMIN_USER},
)
if not result.fetchone():
conn.execute(
text(
"INSERT INTO users (uuid, username, password_hash, role, must_change_password) "
"INSERT OR IGNORE INTO users (uuid, username, password_hash, role, must_change_password) "
"VALUES (:uuid, :u, :p, :r, :m)"
),
{

View File

@@ -6,7 +6,6 @@ forwards events as JSON to LOG_TARGET if set.
"""
import os
import sys
from pathlib import Path
from twisted.internet import defer, reactor

View File

@@ -18,7 +18,7 @@ NODE_NAME = os.environ.get("NODE_NAME", "mailserver")
SERVICE_NAME = "imap"
LOG_TARGET = os.environ.get("LOG_TARGET", "")
PORT = int(os.environ.get("PORT", "143"))
IMAP_BANNER = os.environ.get("IMAP_BANNER", f"* OK Dovecot ready.\r\n")
IMAP_BANNER = os.environ.get("IMAP_BANNER", "* OK Dovecot ready.\r\n")
_RAW_USERS = os.environ.get("IMAP_USERS", "admin:admin123,root:toor,mail:mail,user:user")
VALID_USERS: dict[str, str] = {

View File

@@ -7,7 +7,6 @@ LOG_TARGET if set.
"""
import os
import sys
from twisted.internet import protocol, reactor
from twisted.python import log as twisted_log

View File

@@ -83,7 +83,7 @@ def assert_rfc5424(
criteria = {"service": service, "event_type": event_type, **fields}
raise AssertionError(
f"No RFC 5424 line matching {criteria!r} found among {len(lines)} lines:\n"
+ "\n".join(f" {l!r}" for l in lines[:20])
+ "\n".join(f" {line!r}" for line in lines[:20])
)

View File

@@ -35,5 +35,5 @@ class TestFTPLive:
ftp.close()
lines = drain()
# At least one RFC 5424 line from the ftp service
rfc_lines = [l for l in lines if "<" in l and ">1 " in l and "ftp" in l]
assert rfc_lines, f"No ftp RFC 5424 lines found. stdout:\n" + "\n".join(lines[:15])
rfc_lines = [line for line in lines if "<" in line and ">1 " in line and "ftp" in line]
assert rfc_lines, "No ftp RFC 5424 lines found. stdout:\n" + "\n".join(lines[:15])

View File

@@ -28,8 +28,8 @@ class TestHTTPLive:
)
lines = drain()
# body field present in log line
assert any("body=" in l for l in lines if "request" in l), (
f"Expected 'body=' in request log line. Got:\n" + "\n".join(lines[:10])
assert any("body=" in line for line in lines if "request" in line), (
"Expected 'body=' in request log line. Got:\n" + "\n".join(lines[:10])
)
def test_method_and_path_in_log(self, live_service):

View File

@@ -65,6 +65,6 @@ class TestMongoDBLive:
client.close()
lines = drain()
# At least one message was exchanged
assert any("mongodb" in l for l in lines), (
assert any("mongodb" in line for line in lines), (
"Expected at least one mongodb log line"
)

View File

@@ -8,7 +8,7 @@ length fields that could cause huge buffer allocations.
import importlib.util
import struct
import sys
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock
import pytest
from hypothesis import given, settings

View File

@@ -176,7 +176,7 @@ def test_pop3_list_returns_10_entries(pop3_mod):
resp = _replies(written).decode()
assert resp.startswith("+OK 10")
# Count individual message lines: "N size\r\n"
entries = [l for l in resp.split("\r\n") if l and l[0].isdigit()]
entries = [entry for entry in resp.split("\r\n") if entry and entry[0].isdigit()]
assert len(entries) == 10
@@ -225,7 +225,7 @@ def test_pop3_top_3_body_lines_count(pop3_mod):
parts = resp.split("\r\n\r\n", 1)
assert len(parts) == 2
body_section = parts[1].rstrip("\r\n.")
body_lines = [l for l in body_section.split("\r\n") if l != "."]
body_lines = [part for part in body_section.split("\r\n") if part != "."]
assert len(body_lines) <= 3
@@ -235,7 +235,7 @@ def test_pop3_uidl_returns_10_entries(pop3_mod):
_send(proto, "UIDL")
resp = _replies(written).decode()
assert resp.startswith("+OK")
entries = [l for l in resp.split("\r\n") if l and l[0].isdigit()]
entries = [entry for entry in resp.split("\r\n") if entry and entry[0].isdigit()]
assert len(entries) == 10

View File

@@ -8,7 +8,7 @@ tests for zero/tiny/huge msg_len in both the startup and auth states.
import importlib.util
import struct
import sys
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock
import pytest
from hypothesis import given, settings

View File

@@ -112,8 +112,8 @@ def test_dockerfile_has_rsyslog():
def test_dockerfile_runs_as_root():
lines = [l.strip() for l in _dockerfile_text().splitlines()]
user_lines = [l for l in lines if l.startswith("USER ")]
lines = [line.strip() for line in _dockerfile_text().splitlines()]
user_lines = [line for line in lines if line.startswith("USER ")]
assert user_lines == [], f"Unexpected USER directive(s): {user_lines}"