Split up pycroft.lib.user into separate packages

pycroft/helpers/printing/__init__.py

@@ -100,7 +100,7 @@ def generate_user_sheet(
     new_user: bool,
     wifi: bool,
     bank_account: BankAccount,
-    user: User | None = None,
+    user: User = None,
     user_id: str | None = None,
     plain_user_password: str | None = None,
     generation_purpose: str = "",

pycroft/lib/host.py

@@ -15,7 +15,6 @@
 from pycroft.helpers.net import port_name_sort_key
 from pycroft.lib.logging import log_user_event
 from pycroft.lib.net import get_subnets_for_room, get_free_ip, delete_ip
-from pycroft.lib.user import migrate_user_host
 from pycroft.model.facilities import Room
 from pycroft.model.host import Interface, IP, Host, SwitchPort
 from pycroft.model.port import PatchPort
@@ -95,7 +94,46 @@ def host_edit(host: Host, owner: User, room: Room, name: str, processor: User) -
         host.owner = owner
 
     if host.room != room:
-        migrate_user_host(host, room, processor)
+        migrate_host(session, host, room, processor)
+
+
+def migrate_host(session: Session, host: Host, new_room: Room, processor: User) -> None:
+    """
+    Migrate a Host to a new room and if necessary to a new subnet.
+    If the host changes subnet, it will get a new IP address.
+
+    :param host: Host to be migrated
+    :param new_room: new room of the host
+    :param processor: User processing the migration
+    :return:
+    """
+    old_room = host.room
+    host.room = new_room
+
+    subnets_old = get_subnets_for_room(old_room)
+    subnets = get_subnets_for_room(new_room)
+
+    if subnets_old != subnets:
+        for interface in host.interfaces:
+            old_ips = tuple(ip for ip in interface.ips)
+            for old_ip in old_ips:
+                ip_address, subnet = get_free_ip(subnets)
+                new_ip = IP(interface=interface, address=ip_address, subnet=subnet)
+                session.add(new_ip)
+
+                old_address = old_ip.address
+                session.delete(old_ip)
+
+                message = deferred_gettext("Changed IP of {mac} from {old_ip} to {new_ip}.").format(
+                    old_ip=str(old_address), new_ip=str(new_ip.address), mac=interface.mac
+                )
+                log_user_event(author=processor, user=host.owner, message=message.to_json())
+
+    message = deferred_gettext("Moved host '{name}' from {room_old} to {room_new}.").format(
+        name=host.name, room_old=old_room.short_name, room_new=new_room.short_name
+    )
+
+    log_user_event(author=processor, user=host.owner, message=message.to_json())
 
 
 @with_transaction
@@ -234,3 +272,13 @@ def get_conflicting_interface(
     if new_mac == current_mac:
         return None
     return session.scalar(select(Interface).filter_by(mac=new_mac))
+
+
+def setup_ipv4_networking(session: Session, host: Host) -> None:
+    """Add suitable ips for every interface of a host"""
+    subnets = get_subnets_for_room(host.room)
+
+    for interface in host.interfaces:
+        ip_address, subnet = get_free_ip(subnets)
+        new_ip = IP(interface=interface, address=ip_address, subnet=subnet)
+        session.add(new_ip)

pycroft/lib/mail.py

@@ -2,45 +2,35 @@
 pycroft.lib.mail
 ~~~~~~~~~~~~~~~~
 """
+
+from __future__ import annotations
 import logging
 import os
 import smtplib
 import ssl
 import traceback
 import typing as t
-from dataclasses import dataclass
+from contextvars import ContextVar
+from dataclasses import dataclass, field, InitVar
 from email.header import Header
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
 from email.utils import make_msgid, formatdate
+from functools import lru_cache
 
 import jinja2
+from werkzeug.local import LocalProxy
 
 from pycroft.lib.exc import PycroftLibException
 
-mail_envelope_from = os.environ.get('PYCROFT_MAIL_ENVELOPE_FROM')
-mail_from = os.environ.get('PYCROFT_MAIL_FROM')
-mail_reply_to = os.environ.get('PYCROFT_MAIL_REPLY_TO')
-smtp_host = os.environ.get('PYCROFT_SMTP_HOST')
-smtp_port = int(os.environ.get('PYCROFT_SMTP_PORT', 465))
-smtp_user = os.environ.get('PYCROFT_SMTP_USER')
-smtp_password = os.environ.get('PYCROFT_SMTP_PASSWORD')
-smtp_ssl = os.environ.get('PYCROFT_SMTP_SSL', 'ssl')
-template_path_type = os.environ.get('PYCROFT_TEMPLATE_PATH_TYPE', 'filesystem')
-template_path = os.environ.get('PYCROFT_TEMPLATE_PATH', 'pycroft/templates')
+
+# TODO proxy and DI; set at app init
+_config_var: ContextVar[MailConfig] = ContextVar("config")
+config: MailConfig = LocalProxy(_config_var)  # type: ignore[assignment]
 
 logger = logging.getLogger('mail')
 logger.setLevel(logging.INFO)
 
-template_loader: jinja2.BaseLoader
-if template_path_type == 'filesystem':
-    template_loader = jinja2.FileSystemLoader(searchpath=f'{template_path}/mail')
-else:
-    template_loader = jinja2.PackageLoader(package_name='pycroft',
-                                           package_path=f'{template_path}/mail')
-
-template_env = jinja2.Environment(loader=template_loader)
-
 
 @dataclass
 class Mail:
@@ -51,6 +41,16 @@ class Mail:
     body_html: str | None = None
     reply_to: str | None = None
 
+    @property
+    def body_plain_mime(self) -> MIMEText:
+        return MIMEText(self.body_plain, "plain", _charset="utf-8")
+
+    @property
+    def body_html_mime(self) -> MIMEText | None:
+        if not self.body_html:
+            return None
+        return MIMEText(self.body_html, "html", _charset="utf-8")
+
 
 class MailTemplate:
     template: str
@@ -61,27 +61,35 @@ def __init__(self, **kwargs: t.Any) -> None:
         self.args = kwargs
 
     def render(self, **kwargs: t.Any) -> tuple[str, str]:
-        plain = template_env.get_template(self.template).render(mode='plain', **self.args, **kwargs)
-        html = template_env.get_template(self.template).render(mode='html', **self.args, **kwargs)
+        plain = self.jinja_template.render(mode="plain", **self.args, **kwargs)
+        html = self.jinja_template.render(mode="html", **self.args, **kwargs)
 
         return plain, html
 
+    @property
+    def jinja_template(self) -> jinja2.Template:
+        return _get_template(self.template)
 
-def compose_mail(mail: Mail) -> MIMEMultipart:
-    msg = MIMEMultipart('alternative', _charset='utf-8')
-    msg['Message-Id'] = make_msgid()
-    msg['From'] = mail_from
-    msg['To'] = Header(mail.to_address)
-    msg['Subject'] = mail.subject
-    msg['Date'] = formatdate(localtime=True)
 
-    msg.attach(MIMEText(mail.body_plain, 'plain', _charset='utf-8'))
+@lru_cache(maxsize=None)
+def _get_template(template_location: str) -> jinja2.Template:
+    if config is None:
+        raise RuntimeError("`mail.config` not set up!")
+    return config.template_env.get_template(template_location)
 
-    if mail.body_html is not None:
-        msg.attach(MIMEText(mail.body_html, 'html', _charset='utf-8'))
 
-    if mail.reply_to is not None or mail.reply_to is not None:
-        msg['Reply-To'] = mail_reply_to if mail.reply_to is None else mail.reply_to
+def compose_mail(mail: Mail, from_: str, default_reply_to: str | None) -> MIMEMultipart:
+    msg = MIMEMultipart("alternative", _charset="utf-8")
+    msg["Message-Id"] = make_msgid()
+    msg["From"] = from_
+    msg["To"] = str(Header(mail.to_address))
+    msg["Subject"] = mail.subject
+    msg["Date"] = formatdate(localtime=True)
+    msg.attach(mail.body_plain_mime)
+    if (html := mail.body_html_mime) is not None:
+        msg.attach(html)
+    if reply_to := mail.reply_to or default_reply_to:
+        msg["Reply-To"] = reply_to
 
     print(msg)
 
@@ -100,12 +108,19 @@ def send_mails(mails: list[Mail]) -> tuple[bool, int]:
     :param mails: A list of mails
 
     :returns: Whether the transmission succeeded
+    :context: config
     """
-
-    if not smtp_host:
-        logger.critical("No mailserver config available")
-
-        raise RuntimeError
+    if config is None:
+        raise RuntimeError("`mail.config` not set up!")
+
+    mail_envelope_from = config.mail_envelope_from
+    mail_from = config.mail_from
+    mail_reply_to = config.mail_reply_to
+    smtp_host = config.smtp_host
+    smtp_port = config.smtp_port
+    smtp_user = config.smtp_user
+    smtp_password = config.smtp_password
+    smtp_ssl = config.smtp_ssl
 
     use_ssl = smtp_ssl == 'ssl'
     use_starttls = smtp_ssl == 'starttls'
@@ -159,7 +174,7 @@ def send_mails(mails: list[Mail]) -> tuple[bool, int]:
 
         for mail in mails:
             try:
-                mime_mail = compose_mail(mail)
+                mime_mail = compose_mail(mail, from_=mail_from, default_reply_to=mail_reply_to)
                 assert mail_envelope_from is not None
                 smtp.sendmail(from_addr=mail_envelope_from, to_addrs=mail.to_address,
                               msg=mime_mail.as_string())
@@ -248,3 +263,55 @@ def send_template_mails(
     from pycroft.task import send_mails_async
 
     send_mails_async.delay(mails)
+
+
+@dataclass
+class MailConfig:
+    mail_envelope_from: str
+    mail_from: str
+    mail_reply_to: str | None
+    smtp_host: str
+    smtp_user: str
+    smtp_password: str
+    smtp_port: int = field(default=465)
+    smtp_ssl: str = field(default="ssl")
+
+    template_path_type: InitVar[str | None] = None
+    template_path: InitVar[str | None] = None
+    template_env: jinja2.Environment = field(init=False)
+
+    @classmethod
+    def from_env(cls) -> t.Self:
+        env = os.environ
+        config = cls(
+            mail_envelope_from=env["PYCROFT_MAIL_ENVELOPE_FROM"],
+            mail_from=env["PYCROFT_MAIL_FROM"],
+            mail_reply_to=env.get("PYCROFT_MAIL_REPLY_TO"),
+            smtp_host=env["PYCROFT_SMTP_HOST"],
+            smtp_user=env["PYCROFT_SMTP_USER"],
+            smtp_password=env["PYCROFT_SMTP_PASSWORD"],
+            template_path_type=env.get("PYCROFT_TEMPLATE_PATH_TYPE"),
+            template_path=env.get("PYCROFT_TEMPLATE_PATH"),
+        )
+        if (smtp_port := env.get("PYCROFT_SMTP_PORT")) is not None:
+            config.smtp_port = int(smtp_port)
+        if (smtp_ssl := env.get("PYCROFT_SMTP_SSL")) is not None:
+            config.smtp_ssl = smtp_ssl
+
+        return config
+
+    def __post_init__(self, template_path_type: str | None, template_path: str | None) -> None:
+        template_loader: jinja2.BaseLoader
+        if template_path_type is None:
+            template_path_type = "filesystem"
+        if template_path is None:
+            template_path = "pycroft/templates"
+
+        if template_path_type == "filesystem":
+            template_loader = jinja2.FileSystemLoader(searchpath=f"{template_path}/mail")
+        else:
+            template_loader = jinja2.PackageLoader(
+                package_name="pycroft", package_path=f"{template_path}/mail"
+            )
+
+        self.template_env = jinja2.Environment(loader=template_loader)

pycroft/lib/task.py

@@ -114,12 +114,13 @@ class UserMoveTaskImpl(UserTaskImpl[UserMoveParams]):
 
     def _execute(self, task: UserTask, parameters: UserMoveParams) -> None:
         from pycroft.lib import user as lib_user
+        from pycroft.lib.facilities import get_room
         if task.user.room is None:
             self.errors.append("Tried to move in user, "
                                "but user was already living in a dormitory.")
             return
 
-        room = lib_user.get_room(
+        room = get_room(
             room_number=parameters.room_number,
             level=parameters.level,
             building_id=parameters.building_id,
@@ -147,13 +148,14 @@ class UserMoveInTaskImpl(UserTaskImpl):
 
     def _execute(self, task: UserTask, parameters: UserMoveInParams) -> None:
         from pycroft.lib import user as lib_user
+        from pycroft.lib.facilities import get_room
 
         if task.user.room is not None:
             self.errors.append("Tried to move in user, "
                                "but user was already living in a dormitory.")
             return
 
-        room = lib_user.get_room(
+        room = get_room(
             room_number=parameters.room_number,
             level=parameters.level,
             building_id=parameters.building_id,
@@ -195,7 +197,7 @@ def schedule_user_task(
     due: DateTimeTz,
     user: User,
     parameters: TaskParams,
-    processor: User,
+    processor: User | None,
 ) -> UserTask:
     if due < session.utcnow():
         raise ValueError("the due date must be in the future")