diff --git a/selfprivacy_api/backup/__init__.py b/selfprivacy_api/backup/__init__.py
index 2a068a8..ef07171 100644
--- a/selfprivacy_api/backup/__init__.py
+++ b/selfprivacy_api/backup/__init__.py
@@ -246,7 +246,7 @@ class Backups:
try:
if service.can_be_backed_up() is False:
raise ValueError("cannot backup a non-backuppable service")
- folders = service.get_folders()
+ folders = service.get_folders_to_back_up()
service_name = service.get_id()
service.pre_backup(job=job)
Jobs.update(job, status=JobStatus.RUNNING, status_text="Uploading backup")
@@ -517,7 +517,7 @@ class Backups:
snapshot_id: str,
verify=True,
) -> None:
- folders = service.get_folders()
+ folders = service.get_folders_to_back_up()
Backups.provider().backupper.restore_from_backup(
snapshot_id,
@@ -717,7 +717,7 @@ class Backups:
Returns the amount of space available on the volume the given
service is located on.
"""
- folders = service.get_folders()
+ folders = service.get_folders_to_back_up()
if folders == []:
raise ValueError("unallocated service", service.get_id())
diff --git a/selfprivacy_api/backup/backuppers/restic_backupper.py b/selfprivacy_api/backup/backuppers/restic_backupper.py
index e6a30b4..12df3a6 100644
--- a/selfprivacy_api/backup/backuppers/restic_backupper.py
+++ b/selfprivacy_api/backup/backuppers/restic_backupper.py
@@ -222,11 +222,6 @@ class ResticBackupper(AbstractBackupper):
tags=tags,
)
- logger.warning(
- "Starting backup: " + " ".join(self._censor_command(backup_command))
- )
- logger.warning("Folders: " + str(folders))
-
try:
messages = ResticBackupper._run_backup_command(backup_command, job)
diff --git a/selfprivacy_api/graphql/common_types/service.py b/selfprivacy_api/graphql/common_types/service.py
index 0cc39d2..cd2329f 100644
--- a/selfprivacy_api/graphql/common_types/service.py
+++ b/selfprivacy_api/graphql/common_types/service.py
@@ -6,6 +6,7 @@ import strawberry
from selfprivacy_api.graphql.common_types.backup import BackupReason
from selfprivacy_api.graphql.common_types.dns import DnsRecord
+from selfprivacy_api.models.services import License
from selfprivacy_api.services import ServiceManager
from selfprivacy_api.services import Service as ServiceInterface
from selfprivacy_api.services import ServiceDnsRecord
@@ -71,6 +72,28 @@ class ServiceStatusEnum(Enum):
OFF = "OFF"
+@strawberry.enum
+class SupportLevelEnum(Enum):
+ """Enum representing the support level of a service."""
+
+ NORMAL = "normal"
+ EXPERIMENTAL = "experimental"
+ DEPRECATED = "deprecated"
+ COMMUNITY = "community"
+ UNKNOWN = "unknown"
+
+
+@strawberry.experimental.pydantic.type(model=License)
+class LicenseType:
+ free: strawberry.auto
+ full_name: strawberry.auto
+ redistributable: strawberry.auto
+ short_name: strawberry.auto
+ spdx_id: strawberry.auto
+ url: strawberry.auto
+ deprecated: strawberry.auto
+
+
def get_storage_usage(root: "Service") -> ServiceStorageUsage:
"""Get storage usage for a service"""
service = ServiceManager.get_service_by_id(root.id)
@@ -176,10 +199,15 @@ class Service:
is_required: bool
is_enabled: bool
is_installed: bool
+ is_system_service: bool
can_be_backed_up: bool
backup_description: str
status: ServiceStatusEnum
url: Optional[str]
+ license: List[LicenseType]
+ homepage: Optional[str]
+ source_page: Optional[str]
+ support_level: SupportLevelEnum
@strawberry.field
def dns_records(self) -> Optional[List[DnsRecord]]:
@@ -241,6 +269,13 @@ def service_to_graphql_service(service: ServiceInterface) -> Service:
backup_description=service.get_backup_description(),
status=ServiceStatusEnum(service.get_status().value),
url=service.get_url(),
+ is_system_service=service.is_system_service(),
+ license=[
+ LicenseType.from_pydantic(license) for license in service.get_license()
+ ],
+ homepage=service.get_homepage(),
+ source_page=service.get_source_page(),
+ support_level=SupportLevelEnum(service.get_support_level().value),
)
diff --git a/selfprivacy_api/graphql/queries/backup.py b/selfprivacy_api/graphql/queries/backup.py
index 7eba72d..d018f88 100644
--- a/selfprivacy_api/graphql/queries/backup.py
+++ b/selfprivacy_api/graphql/queries/backup.py
@@ -13,6 +13,7 @@ from selfprivacy_api.graphql.common_types.service import (
Service,
ServiceStatusEnum,
SnapshotInfo,
+ SupportLevelEnum,
service_to_graphql_service,
)
from selfprivacy_api.graphql.common_types.backup import AutobackupQuotas
@@ -53,6 +54,11 @@ def tombstone_service(service_id: str) -> Service:
can_be_backed_up=False,
backup_description="",
is_installed=False,
+ homepage=None,
+ source_page=None,
+ license=[],
+ is_system_service=False,
+ support_level=SupportLevelEnum.UNKNOWN,
)
diff --git a/selfprivacy_api/jobs/migrate_to_binds.py b/selfprivacy_api/jobs/migrate_to_binds.py
index bde60fb..a41dffa 100644
--- a/selfprivacy_api/jobs/migrate_to_binds.py
+++ b/selfprivacy_api/jobs/migrate_to_binds.py
@@ -7,11 +7,8 @@ import logging
from pydantic import BaseModel
from selfprivacy_api.jobs import Job, JobStatus, Jobs
-from selfprivacy_api.services.bitwarden import Bitwarden
-from selfprivacy_api.services.forgejo import Forgejo
+from selfprivacy_api.services import ServiceManager
from selfprivacy_api.services.mailserver import MailServer
-from selfprivacy_api.services.nextcloud import Nextcloud
-from selfprivacy_api.services.pleroma import Pleroma
from selfprivacy_api.utils import ReadUserData, WriteUserData
from selfprivacy_api.utils.huey import huey
from selfprivacy_api.utils.block_devices import BlockDevices
@@ -105,6 +102,50 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
)
return
+ Jobs.update(
+ job=job,
+ status=JobStatus.RUNNING,
+ progress=0,
+ status_text="Checking if services are present.",
+ )
+
+ nextcloud_service = ServiceManager.get_service_by_id("nextcloud")
+ bitwarden_service = ServiceManager.get_service_by_id("bitwarden")
+ gitea_service = ServiceManager.get_service_by_id("gitea")
+ pleroma_service = ServiceManager.get_service_by_id("pleroma")
+
+ if not nextcloud_service:
+ Jobs.update(
+ job=job,
+ status=JobStatus.ERROR,
+ error="Nextcloud service not found.",
+ )
+ return
+
+ if not bitwarden_service:
+ Jobs.update(
+ job=job,
+ status=JobStatus.ERROR,
+ error="Bitwarden service not found.",
+ )
+ return
+
+ if not gitea_service:
+ Jobs.update(
+ job=job,
+ status=JobStatus.ERROR,
+ error="Gitea service not found.",
+ )
+ return
+
+ if not pleroma_service:
+ Jobs.update(
+ job=job,
+ status=JobStatus.ERROR,
+ error="Pleroma service not found.",
+ )
+ return
+
Jobs.update(
job=job,
status=JobStatus.RUNNING,
@@ -172,7 +213,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
status_text="Migrating Nextcloud.",
)
- Nextcloud().stop()
+ nextcloud_service.stop()
# If /volumes/sda1/nextcloud or /volumes/sdb/nextcloud exists, skip it.
if not pathlib.Path("/volumes/sda1/nextcloud").exists():
@@ -187,7 +228,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
)
# Start Nextcloud
- Nextcloud().start()
+ nextcloud_service.start()
# Perform migration of Bitwarden
@@ -198,7 +239,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
status_text="Migrating Bitwarden.",
)
- Bitwarden().stop()
+ bitwarden_service.stop()
if not pathlib.Path("/volumes/sda1/bitwarden").exists():
if not pathlib.Path("/volumes/sdb/bitwarden").exists():
@@ -223,7 +264,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
)
# Start Bitwarden
- Bitwarden().start()
+ bitwarden_service.start()
# Perform migration of Gitea
@@ -234,7 +275,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
status_text="Migrating Gitea.",
)
- Forgejo().stop()
+ gitea_service.stop()
if not pathlib.Path("/volumes/sda1/gitea").exists():
if not pathlib.Path("/volumes/sdb/gitea").exists():
@@ -245,7 +286,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
group="gitea",
)
- Forgejo().start()
+ gitea_service.start()
# Perform migration of Mail server
@@ -287,7 +328,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
status_text="Migrating Pleroma.",
)
- Pleroma().stop()
+ pleroma_service.stop()
if not pathlib.Path("/volumes/sda1/pleroma").exists():
if not pathlib.Path("/volumes/sdb/pleroma").exists():
@@ -311,7 +352,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
group="postgres",
)
- Pleroma().start()
+ pleroma_service.start()
Jobs.update(
job=job,
diff --git a/selfprivacy_api/models/services.py b/selfprivacy_api/models/services.py
index ea9b9dd..1e18bdc 100644
--- a/selfprivacy_api/models/services.py
+++ b/selfprivacy_api/models/services.py
@@ -1,6 +1,16 @@
from enum import Enum
-from typing import Optional
-from pydantic import BaseModel
+from typing import Optional, List
+from pydantic import BaseModel, ConfigDict
+from pydantic.alias_generators import to_camel
+from selfprivacy_api.services.owned_path import OwnedPath
+
+
+class BaseSchema(BaseModel):
+ model_config = ConfigDict(
+ alias_generator=to_camel,
+ populate_by_name=True,
+ from_attributes=True,
+ )
class ServiceStatus(Enum):
@@ -15,6 +25,29 @@ class ServiceStatus(Enum):
OFF = "OFF"
+class SupportLevel(Enum):
+ """Enum representing the support level of a service."""
+
+ NORMAL = "normal"
+ EXPERIMENTAL = "experimental"
+ DEPRECATED = "deprecated"
+ COMMUNITY = "community"
+ UNKNOWN = "unknown"
+
+ @classmethod
+ def from_str(cls, support_level: str) -> "SupportLevel":
+ """Return the SupportLevel from a string."""
+ if support_level == "normal":
+ return cls.NORMAL
+ if support_level == "experimental":
+ return cls.EXPERIMENTAL
+ if support_level == "deprecated":
+ return cls.DEPRECATED
+ if support_level == "community":
+ return cls.COMMUNITY
+ return cls.UNKNOWN
+
+
class ServiceDnsRecord(BaseModel):
type: str
name: str
@@ -23,3 +56,40 @@ class ServiceDnsRecord(BaseModel):
ttl: int
display_name: str
priority: Optional[int] = None
+
+
+class License(BaseSchema):
+ """Model representing a license."""
+
+ deprecated: bool
+ free: bool
+ full_name: str
+ redistributable: bool
+ short_name: str
+ spdx_id: str
+ url: str
+
+
+class ServiceMetaData(BaseSchema):
+ """Model representing the meta data of a service."""
+
+ id: str
+ name: str
+ description: str = "No description found!"
+ svg_icon: str = ""
+ showUrl: bool = True
+ primary_subdomain: Optional[str] = None
+ is_movable: bool = False
+ is_required: bool = False
+ can_be_backed_up: bool = True
+ backup_description: str = "No backup description found!"
+ systemd_services: List[str]
+ user: Optional[str] = None
+ group: Optional[str] = None
+ folders: List[str] = []
+ owned_folders: List[OwnedPath] = []
+ postgre_databases: List[str] = []
+ license: List[License] = []
+ homepage: Optional[str] = None
+ source_page: Optional[str] = None
+ support_level: SupportLevel = SupportLevel.UNKNOWN
diff --git a/selfprivacy_api/services/__init__.py b/selfprivacy_api/services/__init__.py
index fc5bb6e..a37ec7c 100644
--- a/selfprivacy_api/services/__init__.py
+++ b/selfprivacy_api/services/__init__.py
@@ -6,21 +6,14 @@ import typing
import subprocess
import json
from typing import List
-from os import path, remove
+from os import path
from os import makedirs
from os import listdir
from os.path import join
from shutil import copyfile, copytree, rmtree
-from selfprivacy_api.services.bitwarden import Bitwarden
-from selfprivacy_api.services.forgejo import Forgejo
-from selfprivacy_api.services.jitsimeet import JitsiMeet
from selfprivacy_api.services.prometheus import Prometheus
-from selfprivacy_api.services.roundcube import Roundcube
from selfprivacy_api.services.mailserver import MailServer
-from selfprivacy_api.services.nextcloud import Nextcloud
-from selfprivacy_api.services.pleroma import Pleroma
-from selfprivacy_api.services.ocserv import Ocserv
from selfprivacy_api.services.service import Service, ServiceDnsRecord
from selfprivacy_api.services.service import ServiceStatus
@@ -28,7 +21,7 @@ from selfprivacy_api.utils.cached_call import redis_cached_call
import selfprivacy_api.utils.network as network_utils
from selfprivacy_api.services.api_icon import API_ICON
-from selfprivacy_api.utils import USERDATA_FILE, DKIM_DIR, SECRETS_FILE, get_domain
+from selfprivacy_api.utils import USERDATA_FILE, DKIM_DIR, SECRETS_FILE
from selfprivacy_api.utils.block_devices import BlockDevices
from selfprivacy_api.utils import read_account_uri
from selfprivacy_api.services.templated_service import (
@@ -143,6 +136,10 @@ class ServiceManager(Service):
def is_enabled() -> bool:
return True
+ @staticmethod
+ def is_system_service() -> bool:
+ return True
+
@staticmethod
def get_backup_description() -> str:
return "General server settings."
@@ -263,8 +260,6 @@ def get_templated_service(service_id: str) -> TemplatedService:
@redis_cached_call(ttl=3600)
def get_remote_service(id: str, url: str) -> TemplatedService:
- # Get JSON from calling the sp-fetch-remote-module command with the URL
- # Parse the JSON into a TemplatedService object
response = subprocess.run(
["sp-fetch-remote-module", url],
capture_output=True,
@@ -274,23 +269,26 @@ def get_remote_service(id: str, url: str) -> TemplatedService:
return TemplatedService(id, response.stdout)
+DUMMY_SERVICES = []
+TEST_FLAGS: list[str] = []
+
+
@redis_cached_call(ttl=5)
def get_services() -> List[Service]:
+ if "ONLY_DUMMY_SERVICE" in TEST_FLAGS:
+ return DUMMY_SERVICES
+ if "DUMMY_SERVICE_AND_API" in TEST_FLAGS:
+ return DUMMY_SERVICES + [ServiceManager()]
+
hardcoded_services: list[Service] = [
- Bitwarden(),
- # Forgejo(),
MailServer(),
- Nextcloud(),
- # Pleroma(),
- Ocserv(),
- JitsiMeet(),
- Roundcube(),
ServiceManager(),
Prometheus(),
]
+ if DUMMY_SERVICES:
+ hardcoded_services += DUMMY_SERVICES
service_ids = [service.get_id() for service in hardcoded_services]
- # Load services from SP_MODULES_DEFENITIONS_PATH
templated_services: List[Service] = []
if path.exists(SP_MODULES_DEFENITIONS_PATH):
for module in listdir(SP_MODULES_DEFENITIONS_PATH):
@@ -322,6 +320,3 @@ def get_services() -> List[Service]:
logger.error(f"Failed to load service {module}: {e}")
return hardcoded_services + templated_services
-
-
-# services = get_services()
diff --git a/selfprivacy_api/services/bitwarden/__init__.py b/selfprivacy_api/services/bitwarden/__init__.py
deleted file mode 100644
index fe5a018..0000000
--- a/selfprivacy_api/services/bitwarden/__init__.py
+++ /dev/null
@@ -1,109 +0,0 @@
-"""Class representing Bitwarden service"""
-
-import base64
-import subprocess
-from typing import List
-
-from selfprivacy_api.utils.systemd import get_service_status
-from selfprivacy_api.services.service import Service, ServiceStatus
-from selfprivacy_api.services.bitwarden.icon import BITWARDEN_ICON
-from selfprivacy_api.services.config_item import (
- StringServiceConfigItem,
- BoolServiceConfigItem,
- ServiceConfigItem,
-)
-from selfprivacy_api.utils.regex_strings import SUBDOMAIN_REGEX
-
-
-class Bitwarden(Service):
- """Class representing Bitwarden service."""
-
- config_items: dict[str, ServiceConfigItem] = {
- "subdomain": StringServiceConfigItem(
- id="subdomain",
- default_value="password",
- description="Subdomain",
- regex=SUBDOMAIN_REGEX,
- widget="subdomain",
- ),
- "signupsAllowed": BoolServiceConfigItem(
- id="signupsAllowed",
- default_value=True,
- description="Allow new user signups",
- ),
- "sendsAllowed": BoolServiceConfigItem(
- id="sendsAllowed",
- default_value=True,
- description="Allow users to use Bitwarden Send",
- ),
- "emergencyAccessAllowed": BoolServiceConfigItem(
- id="emergencyAccessAllowed",
- default_value=True,
- description="Allow users to enable Emergency Access",
- ),
- }
-
- @staticmethod
- def get_id() -> str:
- """Return service id."""
- return "bitwarden"
-
- @staticmethod
- def get_display_name() -> str:
- """Return service display name."""
- return "Bitwarden"
-
- @staticmethod
- def get_description() -> str:
- """Return service description."""
- return "Bitwarden is a password manager."
-
- @staticmethod
- def get_svg_icon() -> str:
- """Read SVG icon from file and return it as base64 encoded string."""
- return base64.b64encode(BITWARDEN_ICON.encode("utf-8")).decode("utf-8")
-
- @staticmethod
- def get_user() -> str:
- return "vaultwarden"
-
- @staticmethod
- def is_movable() -> bool:
- return True
-
- @staticmethod
- def is_required() -> bool:
- return False
-
- @staticmethod
- def get_backup_description() -> str:
- return "Password database, encryption certificate and attachments."
-
- @staticmethod
- def get_status() -> ServiceStatus:
- """
- Return Bitwarden status from systemd.
- Use command return code to determine status.
-
- Return code 0 means service is running.
- Return code 1 or 2 means service is in error stat.
- Return code 3 means service is stopped.
- Return code 4 means service is off.
- """
- return get_service_status("vaultwarden.service")
-
- @staticmethod
- def stop():
- subprocess.run(["systemctl", "stop", "vaultwarden.service"])
-
- @staticmethod
- def start():
- subprocess.run(["systemctl", "start", "vaultwarden.service"])
-
- @staticmethod
- def restart():
- subprocess.run(["systemctl", "restart", "vaultwarden.service"])
-
- @staticmethod
- def get_folders() -> List[str]:
- return ["/var/lib/bitwarden", "/var/lib/bitwarden_rs"]
diff --git a/selfprivacy_api/services/bitwarden/bitwarden.svg b/selfprivacy_api/services/bitwarden/bitwarden.svg
deleted file mode 100644
index ced270c..0000000
--- a/selfprivacy_api/services/bitwarden/bitwarden.svg
+++ /dev/null
@@ -1,3 +0,0 @@
-
diff --git a/selfprivacy_api/services/bitwarden/icon.py b/selfprivacy_api/services/bitwarden/icon.py
deleted file mode 100644
index f9280e0..0000000
--- a/selfprivacy_api/services/bitwarden/icon.py
+++ /dev/null
@@ -1,5 +0,0 @@
-BITWARDEN_ICON = """
-
-"""
diff --git a/selfprivacy_api/services/forgejo/__init__.py b/selfprivacy_api/services/forgejo/__init__.py
deleted file mode 100644
index e2f86b3..0000000
--- a/selfprivacy_api/services/forgejo/__init__.py
+++ /dev/null
@@ -1,134 +0,0 @@
-"""Class representing Bitwarden service"""
-
-import base64
-import subprocess
-from typing import List
-
-from selfprivacy_api.utils import ReadUserData, WriteUserData
-
-from selfprivacy_api.utils.systemd import get_service_status
-from selfprivacy_api.services.service import Service, ServiceStatus
-from selfprivacy_api.services.forgejo.icon import FORGEJO_ICON
-from selfprivacy_api.services.config_item import (
- StringServiceConfigItem,
- BoolServiceConfigItem,
- EnumServiceConfigItem,
- ServiceConfigItem,
-)
-from selfprivacy_api.utils.regex_strings import SUBDOMAIN_REGEX
-
-
-class Forgejo(Service):
- """Class representing Forgejo service.
-
- Previously was Gitea, so some IDs are still called gitea for compatibility.
- """
-
- config_items: dict[str, ServiceConfigItem] = {
- "subdomain": StringServiceConfigItem(
- id="subdomain",
- default_value="git",
- description="Subdomain",
- regex=SUBDOMAIN_REGEX,
- widget="subdomain",
- ),
- "appName": StringServiceConfigItem(
- id="appName",
- default_value="SelfPrivacy git Service",
- description="The name displayed in the web interface",
- ),
- "enableLfs": BoolServiceConfigItem(
- id="enableLfs",
- default_value=True,
- description="Enable Git LFS",
- ),
- "forcePrivate": BoolServiceConfigItem(
- id="forcePrivate",
- default_value=False,
- description="Force all new repositories to be private",
- ),
- "disableRegistration": BoolServiceConfigItem(
- id="disableRegistration",
- default_value=False,
- description="Disable registration of new users",
- ),
- "requireSigninView": BoolServiceConfigItem(
- id="requireSigninView",
- default_value=False,
- description="Force users to log in to view any page",
- ),
- "defaultTheme": EnumServiceConfigItem(
- id="defaultTheme",
- default_value="forgejo-auto",
- description="Default theme",
- options=[
- "forgejo-auto",
- "forgejo-light",
- "forgejo-dark",
- "gitea-auto",
- "gitea-light",
- "gitea-dark",
- ],
- ),
- }
-
- @staticmethod
- def get_id() -> str:
- """Return service id. For compatibility keep in gitea."""
- return "gitea"
-
- @staticmethod
- def get_display_name() -> str:
- """Return service display name."""
- return "Forgejo"
-
- @staticmethod
- def get_description() -> str:
- """Return service description."""
- return "Forgejo is a Git forge."
-
- @staticmethod
- def get_svg_icon() -> str:
- """Read SVG icon from file and return it as base64 encoded string."""
- return base64.b64encode(FORGEJO_ICON.encode("utf-8")).decode("utf-8")
-
- @staticmethod
- def is_movable() -> bool:
- return True
-
- @staticmethod
- def is_required() -> bool:
- return False
-
- @staticmethod
- def get_backup_description() -> str:
- return "Git repositories, database and user data."
-
- @staticmethod
- def get_status() -> ServiceStatus:
- """
- Return Gitea status from systemd.
- Use command return code to determine status.
- Return code 0 means service is running.
- Return code 1 or 2 means service is in error stat.
- Return code 3 means service is stopped.
- Return code 4 means service is off.
- """
- return get_service_status("forgejo.service")
-
- @staticmethod
- def stop():
- subprocess.run(["systemctl", "stop", "forgejo.service"])
-
- @staticmethod
- def start():
- subprocess.run(["systemctl", "start", "forgejo.service"])
-
- @staticmethod
- def restart():
- subprocess.run(["systemctl", "restart", "forgejo.service"])
-
- @staticmethod
- def get_folders() -> List[str]:
- """The data folder is still called gitea for compatibility."""
- return ["/var/lib/gitea"]
diff --git a/selfprivacy_api/services/forgejo/gitea.svg b/selfprivacy_api/services/forgejo/gitea.svg
deleted file mode 100644
index 9ba8a76..0000000
--- a/selfprivacy_api/services/forgejo/gitea.svg
+++ /dev/null
@@ -1,3 +0,0 @@
-
diff --git a/selfprivacy_api/services/forgejo/icon.py b/selfprivacy_api/services/forgejo/icon.py
deleted file mode 100644
index 5e600cf..0000000
--- a/selfprivacy_api/services/forgejo/icon.py
+++ /dev/null
@@ -1,5 +0,0 @@
-FORGEJO_ICON = """
-
-"""
diff --git a/selfprivacy_api/services/jitsimeet/__init__.py b/selfprivacy_api/services/jitsimeet/__init__.py
deleted file mode 100644
index e317c44..0000000
--- a/selfprivacy_api/services/jitsimeet/__init__.py
+++ /dev/null
@@ -1,109 +0,0 @@
-"""Class representing Jitsi Meet service"""
-
-import base64
-import subprocess
-from typing import List
-
-from selfprivacy_api.jobs import Job
-from selfprivacy_api.utils.systemd import (
- get_service_status_from_several_units,
-)
-from selfprivacy_api.services.service import Service, ServiceStatus
-from selfprivacy_api.utils.block_devices import BlockDevice
-from selfprivacy_api.services.jitsimeet.icon import JITSI_ICON
-from selfprivacy_api.services.config_item import (
- StringServiceConfigItem,
- ServiceConfigItem,
-)
-from selfprivacy_api.utils.regex_strings import SUBDOMAIN_REGEX
-
-
-class JitsiMeet(Service):
- """Class representing Jitsi service"""
-
- config_items: dict[str, ServiceConfigItem] = {
- "subdomain": StringServiceConfigItem(
- id="subdomain",
- default_value="meet",
- description="Subdomain",
- regex=SUBDOMAIN_REGEX,
- widget="subdomain",
- ),
- "appName": StringServiceConfigItem(
- id="appName",
- default_value="Jitsi Meet",
- description="The name displayed in the web interface",
- ),
- }
-
- @staticmethod
- def get_id() -> str:
- """Return service id."""
- return "jitsi-meet"
-
- @staticmethod
- def get_display_name() -> str:
- """Return service display name."""
- return "JitsiMeet"
-
- @staticmethod
- def get_description() -> str:
- """Return service description."""
- return "Jitsi Meet is a free and open-source video conferencing solution."
-
- @staticmethod
- def get_svg_icon() -> str:
- """Read SVG icon from file and return it as base64 encoded string."""
- return base64.b64encode(JITSI_ICON.encode("utf-8")).decode("utf-8")
-
- @staticmethod
- def is_movable() -> bool:
- return False
-
- @staticmethod
- def is_required() -> bool:
- return False
-
- @staticmethod
- def get_backup_description() -> str:
- return "Secrets that are used to encrypt the communication."
-
- @staticmethod
- def get_status() -> ServiceStatus:
- return get_service_status_from_several_units(
- ["prosody.service", "jitsi-videobridge2.service", "jicofo.service"]
- )
-
- @staticmethod
- def stop():
- subprocess.run(
- ["systemctl", "stop", "jitsi-videobridge2.service"],
- check=False,
- )
- subprocess.run(["systemctl", "stop", "jicofo.service"], check=False)
- subprocess.run(["systemctl", "stop", "prosody.service"], check=False)
-
- @staticmethod
- def start():
- subprocess.run(["systemctl", "start", "prosody.service"], check=False)
- subprocess.run(
- ["systemctl", "start", "jitsi-videobridge2.service"],
- check=False,
- )
- subprocess.run(["systemctl", "start", "jicofo.service"], check=False)
-
- @staticmethod
- def restart():
- subprocess.run(["systemctl", "restart", "prosody.service"], check=False)
- subprocess.run(
- ["systemctl", "restart", "jitsi-videobridge2.service"],
- check=False,
- )
- subprocess.run(["systemctl", "restart", "jicofo.service"], check=False)
-
- @staticmethod
- def get_folders() -> List[str]:
- return ["/var/lib/jitsi-meet"]
-
- def move_to_volume(self, volume: BlockDevice) -> Job:
- raise NotImplementedError("jitsi-meet service is not movable")
diff --git a/selfprivacy_api/services/jitsimeet/icon.py b/selfprivacy_api/services/jitsimeet/icon.py
deleted file mode 100644
index 08bcbb1..0000000
--- a/selfprivacy_api/services/jitsimeet/icon.py
+++ /dev/null
@@ -1,5 +0,0 @@
-JITSI_ICON = """
-
-"""
diff --git a/selfprivacy_api/services/nextcloud/__init__.py b/selfprivacy_api/services/nextcloud/__init__.py
deleted file mode 100644
index 49ff6d8..0000000
--- a/selfprivacy_api/services/nextcloud/__init__.py
+++ /dev/null
@@ -1,99 +0,0 @@
-"""Class representing Nextcloud service."""
-
-import base64
-import subprocess
-from typing import List
-
-from selfprivacy_api.utils.systemd import get_service_status
-from selfprivacy_api.services.service import Service, ServiceStatus
-
-from selfprivacy_api.services.nextcloud.icon import NEXTCLOUD_ICON
-from selfprivacy_api.services.config_item import (
- StringServiceConfigItem,
- BoolServiceConfigItem,
- ServiceConfigItem,
-)
-from selfprivacy_api.utils.regex_strings import SUBDOMAIN_REGEX
-
-
-class Nextcloud(Service):
- """Class representing Nextcloud service."""
-
- config_items: dict[str, ServiceConfigItem] = {
- "subdomain": StringServiceConfigItem(
- id="subdomain",
- default_value="cloud",
- description="Subdomain",
- regex=SUBDOMAIN_REGEX,
- widget="subdomain",
- ),
- "enableImagemagick": BoolServiceConfigItem(
- id="enableImagemagick",
- default_value=True,
- description="Enable ImageMagick",
- ),
- }
-
- @staticmethod
- def get_id() -> str:
- """Return service id."""
- return "nextcloud"
-
- @staticmethod
- def get_display_name() -> str:
- """Return service display name."""
- return "Nextcloud"
-
- @staticmethod
- def get_description() -> str:
- """Return service description."""
- return "Nextcloud is a cloud storage service that offers a web interface and a desktop client."
-
- @staticmethod
- def get_svg_icon() -> str:
- """Read SVG icon from file and return it as base64 encoded string."""
- return base64.b64encode(NEXTCLOUD_ICON.encode("utf-8")).decode("utf-8")
-
- @staticmethod
- def is_movable() -> bool:
- return True
-
- @staticmethod
- def is_required() -> bool:
- return False
-
- @staticmethod
- def get_backup_description() -> str:
- return "All the files and other data stored in Nextcloud."
-
- @staticmethod
- def get_status() -> ServiceStatus:
- """
- Return Nextcloud status from systemd.
- Use command return code to determine status.
-
- Return code 0 means service is running.
- Return code 1 or 2 means service is in error stat.
- Return code 3 means service is stopped.
- Return code 4 means service is off.
- """
- return get_service_status("phpfpm-nextcloud.service")
-
- @staticmethod
- def stop():
- """Stop Nextcloud service."""
- subprocess.Popen(["systemctl", "stop", "phpfpm-nextcloud.service"])
-
- @staticmethod
- def start():
- """Start Nextcloud service."""
- subprocess.Popen(["systemctl", "start", "phpfpm-nextcloud.service"])
-
- @staticmethod
- def restart():
- """Restart Nextcloud service."""
- subprocess.Popen(["systemctl", "restart", "phpfpm-nextcloud.service"])
-
- @staticmethod
- def get_folders() -> List[str]:
- return ["/var/lib/nextcloud"]
diff --git a/selfprivacy_api/services/nextcloud/icon.py b/selfprivacy_api/services/nextcloud/icon.py
deleted file mode 100644
index d178640..0000000
--- a/selfprivacy_api/services/nextcloud/icon.py
+++ /dev/null
@@ -1,12 +0,0 @@
-NEXTCLOUD_ICON = """
-
-"""
diff --git a/selfprivacy_api/services/nextcloud/nextcloud.svg b/selfprivacy_api/services/nextcloud/nextcloud.svg
deleted file mode 100644
index d7dbcb5..0000000
--- a/selfprivacy_api/services/nextcloud/nextcloud.svg
+++ /dev/null
@@ -1,10 +0,0 @@
-
diff --git a/selfprivacy_api/services/ocserv/__init__.py b/selfprivacy_api/services/ocserv/__init__.py
deleted file mode 100644
index ae88260..0000000
--- a/selfprivacy_api/services/ocserv/__init__.py
+++ /dev/null
@@ -1,82 +0,0 @@
-"""Class representing ocserv service."""
-
-import base64
-import subprocess
-import typing
-from selfprivacy_api.jobs import Job
-from selfprivacy_api.utils.systemd import get_service_status
-from selfprivacy_api.services.service import Service, ServiceStatus
-from selfprivacy_api.utils.block_devices import BlockDevice
-from selfprivacy_api.services.ocserv.icon import OCSERV_ICON
-
-
-class Ocserv(Service):
- """Class representing ocserv service."""
-
- @staticmethod
- def get_id() -> str:
- return "ocserv"
-
- @staticmethod
- def get_display_name() -> str:
- return "OpenConnect VPN"
-
- @staticmethod
- def get_description() -> str:
- return "OpenConnect VPN to connect your devices and access the internet."
-
- @staticmethod
- def get_svg_icon() -> str:
- return base64.b64encode(OCSERV_ICON.encode("utf-8")).decode("utf-8")
-
- @classmethod
- def get_url(cls) -> typing.Optional[str]:
- """Return service url."""
- return None
-
- @staticmethod
- def is_movable() -> bool:
- return False
-
- @staticmethod
- def is_required() -> bool:
- return False
-
- @staticmethod
- def can_be_backed_up() -> bool:
- return False
-
- @staticmethod
- def get_backup_description() -> str:
- return "Nothing to backup."
-
- @staticmethod
- def get_status() -> ServiceStatus:
- return get_service_status("ocserv.service")
-
- @staticmethod
- def stop():
- subprocess.run(["systemctl", "stop", "ocserv.service"], check=False)
-
- @staticmethod
- def start():
- subprocess.run(["systemctl", "start", "ocserv.service"], check=False)
-
- @staticmethod
- def restart():
- subprocess.run(["systemctl", "restart", "ocserv.service"], check=False)
-
- @classmethod
- def get_configuration(cls):
- return {}
-
- @classmethod
- def set_configuration(cls, config_items):
- return super().set_configuration(config_items)
-
- @staticmethod
- def get_folders() -> typing.List[str]:
- return []
-
- def move_to_volume(self, volume: BlockDevice) -> Job:
- raise NotImplementedError("ocserv service is not movable")
diff --git a/selfprivacy_api/services/ocserv/icon.py b/selfprivacy_api/services/ocserv/icon.py
deleted file mode 100644
index 6585c5e..0000000
--- a/selfprivacy_api/services/ocserv/icon.py
+++ /dev/null
@@ -1,5 +0,0 @@
-OCSERV_ICON = """
-
-"""
diff --git a/selfprivacy_api/services/ocserv/ocserv.svg b/selfprivacy_api/services/ocserv/ocserv.svg
deleted file mode 100644
index 288f743..0000000
--- a/selfprivacy_api/services/ocserv/ocserv.svg
+++ /dev/null
@@ -1,3 +0,0 @@
-
diff --git a/selfprivacy_api/services/pleroma/__init__.py b/selfprivacy_api/services/pleroma/__init__.py
deleted file mode 100644
index ac3fa1d..0000000
--- a/selfprivacy_api/services/pleroma/__init__.py
+++ /dev/null
@@ -1,89 +0,0 @@
-"""Class representing Nextcloud service."""
-
-import base64
-import subprocess
-from typing import List
-
-from selfprivacy_api.services.owned_path import OwnedPath
-from selfprivacy_api.utils.systemd import get_service_status
-from selfprivacy_api.services.service import Service, ServiceStatus
-
-from selfprivacy_api.services.pleroma.icon import PLEROMA_ICON
-
-
-class Pleroma(Service):
- """Class representing Pleroma service."""
-
- @staticmethod
- def get_id() -> str:
- return "pleroma"
-
- @staticmethod
- def get_display_name() -> str:
- return "Pleroma"
-
- @staticmethod
- def get_description() -> str:
- return "Pleroma is a microblogging service that offers a web interface and a desktop client."
-
- @staticmethod
- def get_svg_icon() -> str:
- return base64.b64encode(PLEROMA_ICON.encode("utf-8")).decode("utf-8")
-
- @staticmethod
- def is_movable() -> bool:
- return True
-
- @staticmethod
- def is_required() -> bool:
- return False
-
- @staticmethod
- def get_backup_description() -> str:
- return "Your Pleroma accounts, posts and media."
-
- @staticmethod
- def get_status() -> ServiceStatus:
- return get_service_status("pleroma.service")
-
- @staticmethod
- def stop():
- subprocess.run(["systemctl", "stop", "pleroma.service"])
- subprocess.run(["systemctl", "stop", "postgresql.service"])
-
- @staticmethod
- def start():
- subprocess.run(["systemctl", "start", "pleroma.service"])
- subprocess.run(["systemctl", "start", "postgresql.service"])
-
- @staticmethod
- def restart():
- subprocess.run(["systemctl", "restart", "pleroma.service"])
- subprocess.run(["systemctl", "restart", "postgresql.service"])
-
- @classmethod
- def get_configuration(cls):
- return {}
-
- @classmethod
- def set_configuration(cls, config_items):
- return super().set_configuration(config_items)
-
- @staticmethod
- def get_owned_folders() -> List[OwnedPath]:
- """
- Get a list of occupied directories with ownership info
- Pleroma has folders that are owned by different users
- """
- return [
- OwnedPath(
- path="/var/lib/pleroma",
- owner="pleroma",
- group="pleroma",
- ),
- OwnedPath(
- path="/var/lib/postgresql",
- owner="postgres",
- group="postgres",
- ),
- ]
diff --git a/selfprivacy_api/services/pleroma/icon.py b/selfprivacy_api/services/pleroma/icon.py
deleted file mode 100644
index c0c4d2b..0000000
--- a/selfprivacy_api/services/pleroma/icon.py
+++ /dev/null
@@ -1,12 +0,0 @@
-PLEROMA_ICON = """
-
-"""
diff --git a/selfprivacy_api/services/pleroma/pleroma.svg b/selfprivacy_api/services/pleroma/pleroma.svg
deleted file mode 100644
index f87c438..0000000
--- a/selfprivacy_api/services/pleroma/pleroma.svg
+++ /dev/null
@@ -1,10 +0,0 @@
-
diff --git a/selfprivacy_api/services/prometheus/__init__.py b/selfprivacy_api/services/prometheus/__init__.py
index f095ed1..8b4bf4c 100644
--- a/selfprivacy_api/services/prometheus/__init__.py
+++ b/selfprivacy_api/services/prometheus/__init__.py
@@ -47,6 +47,10 @@ class Prometheus(Service):
def is_required() -> bool:
return True
+ @staticmethod
+ def is_system_service() -> bool:
+ return True
+
@staticmethod
def can_be_backed_up() -> bool:
return False
diff --git a/selfprivacy_api/services/roundcube/__init__.py b/selfprivacy_api/services/roundcube/__init__.py
deleted file mode 100644
index 4b96ed0..0000000
--- a/selfprivacy_api/services/roundcube/__init__.py
+++ /dev/null
@@ -1,100 +0,0 @@
-"""Class representing Roundcube service"""
-
-import base64
-import subprocess
-from typing import List
-
-from selfprivacy_api.jobs import Job
-from selfprivacy_api.utils.systemd import (
- get_service_status_from_several_units,
-)
-from selfprivacy_api.services.service import Service, ServiceStatus
-from selfprivacy_api.utils.block_devices import BlockDevice
-from selfprivacy_api.services.roundcube.icon import ROUNDCUBE_ICON
-from selfprivacy_api.services.config_item import (
- StringServiceConfigItem,
- ServiceConfigItem,
-)
-from selfprivacy_api.utils.regex_strings import SUBDOMAIN_REGEX
-
-
-class Roundcube(Service):
- """Class representing roundcube service"""
-
- config_items: dict[str, ServiceConfigItem] = {
- "subdomain": StringServiceConfigItem(
- id="subdomain",
- default_value="roundcube",
- description="Subdomain",
- regex=SUBDOMAIN_REGEX,
- widget="subdomain",
- ),
- }
-
- @staticmethod
- def get_id() -> str:
- """Return service id."""
- return "roundcube"
-
- @staticmethod
- def get_display_name() -> str:
- """Return service display name."""
- return "Roundcube"
-
- @staticmethod
- def get_description() -> str:
- """Return service description."""
- return "Roundcube is an open source webmail software."
-
- @staticmethod
- def get_svg_icon() -> str:
- """Read SVG icon from file and return it as base64 encoded string."""
- return base64.b64encode(ROUNDCUBE_ICON.encode("utf-8")).decode("utf-8")
-
- @staticmethod
- def is_movable() -> bool:
- return False
-
- @staticmethod
- def is_required() -> bool:
- return False
-
- @staticmethod
- def can_be_backed_up() -> bool:
- return False
-
- @staticmethod
- def get_backup_description() -> str:
- return "Nothing to backup."
-
- @staticmethod
- def get_status() -> ServiceStatus:
- return get_service_status_from_several_units(["phpfpm-roundcube.service"])
-
- @staticmethod
- def stop():
- subprocess.run(
- ["systemctl", "stop", "phpfpm-roundcube.service"],
- check=False,
- )
-
- @staticmethod
- def start():
- subprocess.run(
- ["systemctl", "start", "phpfpm-roundcube.service"],
- check=False,
- )
-
- @staticmethod
- def restart():
- subprocess.run(
- ["systemctl", "restart", "phpfpm-roundcube.service"],
- check=False,
- )
-
- @staticmethod
- def get_folders() -> List[str]:
- return []
-
- def move_to_volume(self, volume: BlockDevice) -> Job:
- raise NotImplementedError("roundcube service is not movable")
diff --git a/selfprivacy_api/services/roundcube/icon.py b/selfprivacy_api/services/roundcube/icon.py
deleted file mode 100644
index 4a08207..0000000
--- a/selfprivacy_api/services/roundcube/icon.py
+++ /dev/null
@@ -1,7 +0,0 @@
-ROUNDCUBE_ICON = """
-
-"""
diff --git a/selfprivacy_api/services/service.py b/selfprivacy_api/services/service.py
index 376a2ac..ca180f3 100644
--- a/selfprivacy_api/services/service.py
+++ b/selfprivacy_api/services/service.py
@@ -15,7 +15,12 @@ from selfprivacy_api.utils.block_devices import BlockDevice, BlockDevices
from selfprivacy_api.jobs import Job, Jobs, JobStatus, report_progress
from selfprivacy_api.jobs.upgrade_system import rebuild_system
-from selfprivacy_api.models.services import ServiceStatus, ServiceDnsRecord
+from selfprivacy_api.models.services import (
+ License,
+ ServiceStatus,
+ ServiceDnsRecord,
+ SupportLevel,
+)
from selfprivacy_api.services.generic_size_counter import get_storage_usage
from selfprivacy_api.services.owned_path import OwnedPath, Bind
from selfprivacy_api.services.moving import (
@@ -166,6 +171,37 @@ class Service(ABC):
with ReadUserData() as user_data:
return user_data.get("modules", {}).get(name, {}) != {}
+ def is_system_service(self) -> bool:
+ """
+ `True` if the service is a system service and should be hidden from the user.
+ `False` if it is not a system service.
+ """
+ return False
+
+ def get_license(self) -> List[License]:
+ """
+ The licenses of the service.
+ """
+ return []
+
+ def get_homepage(self) -> Optional[str]:
+ """
+ The homepage of the service.
+ """
+ return None
+
+ def get_source_page(self) -> Optional[str]:
+ """
+ The source page of the service.
+ """
+ return None
+
+ def get_support_level(self) -> SupportLevel:
+ """
+ The support level of the service.
+ """
+ return SupportLevel.NORMAL
+
@staticmethod
@abstractmethod
def get_status() -> ServiceStatus:
@@ -312,6 +348,10 @@ class Service(ABC):
)
return [owned_folder.path for owned_folder in cls.get_owned_folders()]
+ @classmethod
+ def get_folders_to_back_up(cls) -> List[str]:
+ return cls.get_folders()
+
@classmethod
def get_owned_folders(cls) -> List[OwnedPath]:
"""
diff --git a/selfprivacy_api/services/templated_service.py b/selfprivacy_api/services/templated_service.py
index dbc4762..346e1eb 100644
--- a/selfprivacy_api/services/templated_service.py
+++ b/selfprivacy_api/services/templated_service.py
@@ -1,7 +1,6 @@
"""A Service implementation that loads all needed data from a JSON file"""
import base64
-from enum import Enum
import logging
import json
import subprocess
@@ -9,12 +8,15 @@ from typing import List, Optional
from os.path import join, exists
from os import mkdir, remove
-from pydantic import BaseModel, ConfigDict
-from pydantic.alias_generators import to_camel
-
-from selfprivacy_api.backup.postgres import PostgresDumper
+from selfprivacy_api.utils.postgres import PostgresDumper
from selfprivacy_api.jobs import Job, JobStatus, Jobs
-from selfprivacy_api.models.services import ServiceDnsRecord, ServiceStatus
+from selfprivacy_api.models.services import (
+ License,
+ ServiceDnsRecord,
+ ServiceMetaData,
+ ServiceStatus,
+ SupportLevel,
+)
from selfprivacy_api.services.flake_service_manager import FlakeServiceManager
from selfprivacy_api.services.generic_size_counter import get_storage_usage
from selfprivacy_api.services.owned_path import OwnedPath
@@ -36,26 +38,6 @@ SP_SUGGESTED_MODULES_PATH = "/etc/suggested-sp-modules"
logger = logging.getLogger(__name__)
-class SupportLevel(Enum):
- """Enum representing the support level of a service."""
-
- NORMAL = "normal"
- EXPERIMENTAL = "experimental"
- DEPRECATED = "deprecated"
- UNKNOWN = "unknown"
-
- @classmethod
- def from_str(cls, support_level: str) -> "SupportLevel":
- """Return the SupportLevel from a string."""
- if support_level == "normal":
- return cls.NORMAL
- if support_level == "experimental":
- return cls.EXPERIMENTAL
- if support_level == "deprecated":
- return cls.DEPRECATED
- return cls.UNKNOWN
-
-
def config_item_from_json(json_data: dict) -> Optional[ServiceConfigItem]:
"""Create a ServiceConfigItem from JSON data."""
weight = json_data.get("meta", {}).get("weight", 50)
@@ -103,51 +85,6 @@ def config_item_from_json(json_data: dict) -> Optional[ServiceConfigItem]:
raise ValueError("Unknown config item type")
-class BaseSchema(BaseModel):
- model_config = ConfigDict(
- alias_generator=to_camel,
- populate_by_name=True,
- from_attributes=True,
- )
-
-
-class License(BaseSchema):
- """Model representing a license."""
-
- deprecated: bool
- free: bool
- full_name: str
- redistributable: bool
- short_name: str
- spdx_id: str
- url: str
-
-
-class ServiceMetaData(BaseSchema):
- """Model representing the meta data of a service."""
-
- id: str
- name: str
- description: str = "No description found!"
- svg_icon: str = ""
- showUrl: bool = True
- primary_subdomain: Optional[str] = None
- is_movable: bool = False
- is_required: bool = False
- can_be_backed_up: bool = True
- backup_description: str = "No backup description found!"
- systemd_services: List[str]
- user: Optional[str] = None
- group: Optional[str] = None
- folders: List[str] = []
- owned_folders: List[OwnedPath] = []
- postgre_databases: List[str] = []
- license: List[License] = []
- homepage: Optional[str] = None
- source_page: Optional[str] = None
- support_level: SupportLevel = SupportLevel.UNKNOWN
-
-
class TemplatedService(Service):
"""Class representing a dynamically loaded service."""
@@ -279,6 +216,18 @@ class TemplatedService(Service):
with FlakeServiceManager() as service_manager:
return name in service_manager.services
+ def get_license(self) -> List[License]:
+ return self.meta.license
+
+ def get_homepage(self) -> Optional[str]:
+ return self.meta.homepage
+
+ def get_source_page(self) -> Optional[str]:
+ return self.meta.source_page
+
+ def get_support_level(self) -> SupportLevel:
+ return self.meta.support_level
+
def get_status(self) -> ServiceStatus:
if not self.meta.systemd_services:
return ServiceStatus.INACTIVE
@@ -443,8 +392,6 @@ class TemplatedService(Service):
resulting_folders = folders.copy()
for folder in owned_folders:
resulting_folders.append(folder.path)
- if self.get_postgresql_databases():
- resulting_folders.append(self._get_db_dumps_folder())
return resulting_folders
def get_owned_folders(self) -> List[OwnedPath]:
@@ -453,14 +400,12 @@ class TemplatedService(Service):
resulting_folders = owned_folders.copy()
for folder in folders:
resulting_folders.append(self.owned_path(folder))
+ return resulting_folders
+
+ def get_folders_to_back_up(self) -> List[str]:
+ resulting_folders = self.meta.folders.copy()
if self.get_postgresql_databases():
- resulting_folders.append(
- OwnedPath(
- path=self._get_db_dumps_folder(),
- owner="selfprivacy-api",
- group="selfprivacy-api",
- )
- )
+ resulting_folders.append(self._get_db_dumps_folder())
return resulting_folders
def set_location(self, volume: BlockDevice):
@@ -504,16 +449,13 @@ class TemplatedService(Service):
)
def pre_backup(self, job: Job):
- logger.warning("Pre backup")
if self.get_postgresql_databases():
db_dumps_folder = self._get_db_dumps_folder()
- logger.warning("Pre backup: postgresql databases")
# Create folder for the dumps if it does not exist
if not exists(db_dumps_folder):
mkdir(db_dumps_folder)
# Dump the databases
for db_name in self.get_postgresql_databases():
- logger.warning(f"Pre backup: db_name: {db_name}")
if job is not None:
Jobs.update(
job,
@@ -522,7 +464,6 @@ class TemplatedService(Service):
)
db_dumper = PostgresDumper(db_name)
backup_file = join(db_dumps_folder, f"{db_name}.dump")
- logger.warning(f"Pre backup: backup_file: {backup_file}")
db_dumper.backup_database(backup_file)
def _clear_db_dumps(self):
diff --git a/selfprivacy_api/services/test_service/__init__.py b/selfprivacy_api/services/test_service/__init__.py
index 83d17cc..8f0e50d 100644
--- a/selfprivacy_api/services/test_service/__init__.py
+++ b/selfprivacy_api/services/test_service/__init__.py
@@ -95,7 +95,7 @@ class DummyService(Service):
def get_status(cls) -> ServiceStatus:
filepath = cls.status_file()
if filepath in [None, ""]:
- raise ValueError(f"We do not have a path for our test dummy status file!")
+ raise ValueError("We do not have a path for our test dummy status file!")
if not path.exists(filepath):
raise FileNotFoundError(filepath)
diff --git a/selfprivacy_api/backup/postgres.py b/selfprivacy_api/utils/postgres.py
similarity index 100%
rename from selfprivacy_api/backup/postgres.py
rename to selfprivacy_api/utils/postgres.py
diff --git a/tests/conftest.py b/tests/conftest.py
index 275389c..a9766ed 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -258,7 +258,7 @@ def dummy_service(
ensure_user_exists(user)
# register our service
- services.services.append(service)
+ services.DUMMY_SERVICES.append(service)
huey.immediate = True
assert huey.immediate is True
@@ -269,8 +269,8 @@ def dummy_service(
# Cleanup because apparently it matters wrt tasks
# Some tests may remove it from the list intentionally, this is fine
- if service in services.services:
- services.services.remove(service)
+ if service in services.DUMMY_SERVICES:
+ services.DUMMY_SERVICES.remove(service)
def prepare_nixos_rebuild_calls(fp, unit_name):
diff --git a/tests/test_backup.py b/tests/test_backup.py
index 16589ec..3e23109 100644
--- a/tests/test_backup.py
+++ b/tests/test_backup.py
@@ -250,6 +250,7 @@ def test_error_censoring_encryptionkey(dummy_service, backups):
Backups.back_up(dummy_service)
job = get_backup_fail(dummy_service)
+ assert job is not None
assert_job_errored(job)
job_text = all_job_text(job)
@@ -287,6 +288,7 @@ def test_error_censoring_loginkey(dummy_service, backups, fp):
Backups.back_up(dummy_service)
job = get_backup_fail(dummy_service)
+ assert job is not None
assert_job_errored(job)
job_text = all_job_text(job)
@@ -705,6 +707,7 @@ def test_provider_storage(backups):
Storage.store_provider(test_provider)
restored_provider_model = Storage.load_provider()
+ assert restored_provider_model is not None
assert restored_provider_model.kind == "BACKBLAZE"
assert restored_provider_model.login == test_login
assert restored_provider_model.key == test_key
diff --git a/tests/test_graphql/test_services.py b/tests/test_graphql/test_services.py
index 8d0f0a4..e2754af 100644
--- a/tests/test_graphql/test_services.py
+++ b/tests/test_graphql/test_services.py
@@ -88,20 +88,26 @@ def dummy_service_with_binds(dummy_service, mock_lsblk_devices, volume_folders):
@pytest.fixture()
def only_dummy_service(dummy_service) -> Generator[DummyService, None, None]:
# because queries to services that are not really there error out
- back_copy = service_module.services.copy()
- service_module.services.clear()
- service_module.services.append(dummy_service)
+ service_module.TEST_FLAGS.clear()
+ service_module.TEST_FLAGS.append("ONLY_DUMMY_SERVICE")
+ service_module.DUMMY_SERVICES.clear()
+ service_module.DUMMY_SERVICES.append(dummy_service)
yield dummy_service
- service_module.services.clear()
- service_module.services.extend(back_copy)
+ service_module.TEST_FLAGS.clear()
+ service_module.DUMMY_SERVICES.clear()
@pytest.fixture
def only_dummy_service_and_api(
- only_dummy_service, generic_userdata, dkim_file
+ generic_userdata, dkim_file, dummy_service
) -> Generator[DummyService, None, None]:
- service_module.services.append(ServiceManager())
- return only_dummy_service
+ service_module.TEST_FLAGS.clear()
+ service_module.TEST_FLAGS.append("DUMMY_SERVICE_AND_API")
+ service_module.DUMMY_SERVICES.clear()
+ service_module.DUMMY_SERVICES.append(dummy_service)
+ yield dummy_service
+ service_module.TEST_FLAGS.clear()
+ service_module.DUMMY_SERVICES.clear()
@pytest.fixture()
@@ -727,6 +733,7 @@ def test_graphql_move_service(
def test_mailservice_cannot_enable_disable(authorized_client):
mailservice = ServiceManager.get_service_by_id("simple-nixos-mailserver")
+ assert mailservice is not None
mutation_response = api_enable(authorized_client, mailservice)
data = get_data(mutation_response)["services"]["enableService"]
diff --git a/tests/test_services.py b/tests/test_services.py
index 4c31743..cd4a6bc 100644
--- a/tests/test_services.py
+++ b/tests/test_services.py
@@ -10,8 +10,6 @@ from selfprivacy_api.utils.waitloop import wait_until_true
import selfprivacy_api.services as services_module
-from selfprivacy_api.services.bitwarden import Bitwarden
-from selfprivacy_api.services.pleroma import Pleroma
from selfprivacy_api.services.mailserver import MailServer
from selfprivacy_api.services.owned_path import OwnedPath
@@ -63,22 +61,22 @@ def test_delayed_start_stop(raw_dummy_service):
assert dummy.get_status() == ServiceStatus.ACTIVE
-def test_owned_folders_from_not_owned():
- assert Bitwarden.get_owned_folders() == [
- OwnedPath(
- path=folder,
- group="vaultwarden",
- owner="vaultwarden",
- )
- for folder in Bitwarden.get_folders()
- ]
+# def test_owned_folders_from_not_owned():
+# assert Bitwarden.get_owned_folders() == [
+# OwnedPath(
+# path=folder,
+# group="vaultwarden",
+# owner="vaultwarden",
+# )
+# for folder in Bitwarden.get_folders()
+# ]
-def test_paths_from_owned_paths():
- assert len(Pleroma.get_folders()) == 2
- assert Pleroma.get_folders() == [
- ownedpath.path for ownedpath in Pleroma.get_owned_folders()
- ]
+# def test_paths_from_owned_paths():
+# assert len(Pleroma.get_folders()) == 2
+# assert Pleroma.get_folders() == [
+# ownedpath.path for ownedpath in Pleroma.get_owned_folders()
+# ]
def test_enabling_disabling_reads_json(dummy_service: DummyService):
diff --git a/tests/test_services_systemctl.py b/tests/test_services_systemctl.py
index 43805e8..b83dc9a 100644
--- a/tests/test_services_systemctl.py
+++ b/tests/test_services_systemctl.py
@@ -1,12 +1,7 @@
import pytest
from selfprivacy_api.services.service import ServiceStatus
-from selfprivacy_api.services.bitwarden import Bitwarden
-from selfprivacy_api.services.forgejo import Forgejo
from selfprivacy_api.services.mailserver import MailServer
-from selfprivacy_api.services.nextcloud import Nextcloud
-from selfprivacy_api.services.ocserv import Ocserv
-from selfprivacy_api.services.pleroma import Pleroma
def expected_status_call(service_name: str):
@@ -76,19 +71,9 @@ def mock_popen_systemctl_service_not_ok(mocker):
def test_systemctl_ok(mock_popen_systemctl_service_ok):
assert MailServer.get_status() == ServiceStatus.ACTIVE
- assert Bitwarden.get_status() == ServiceStatus.ACTIVE
- assert Forgejo.get_status() == ServiceStatus.ACTIVE
- assert Nextcloud.get_status() == ServiceStatus.ACTIVE
- assert Ocserv.get_status() == ServiceStatus.ACTIVE
- assert Pleroma.get_status() == ServiceStatus.ACTIVE
call_args_asserts(mock_popen_systemctl_service_ok)
def test_systemctl_failed_service(mock_popen_systemctl_service_not_ok):
assert MailServer.get_status() == ServiceStatus.FAILED
- assert Bitwarden.get_status() == ServiceStatus.FAILED
- assert Forgejo.get_status() == ServiceStatus.FAILED
- assert Nextcloud.get_status() == ServiceStatus.FAILED
- assert Ocserv.get_status() == ServiceStatus.FAILED
- assert Pleroma.get_status() == ServiceStatus.FAILED
call_args_asserts(mock_popen_systemctl_service_not_ok)