feature(root daemon): allow imports in root daemon, even though it is probably a bad idea

This commit is contained in:
Houkime 2024-12-29 16:35:48 +00:00
parent ad1076ad4f
commit 402d2a1303
5 changed files with 49 additions and 17 deletions

View file

@ -0,0 +1 @@
SOCKET_PATH = "/tmp/socket_test.s"

View file

@ -9,7 +9,10 @@ import socket as socket_module
import subprocess import subprocess
from typing import Optional from typing import Optional
SOCKET_PATH = "/tmp/socket_test.s" from . import SOCKET_PATH
# from selfprivacy_api.services.templated_service import SP_MODULES_DEFENITIONS_PATH
BUFFER_SIZE = 1024 BUFFER_SIZE = 1024
services = [ services = [
@ -53,6 +56,14 @@ CHOWN_COMMAND = "chown selfprivacy"
# ] # ]
def sync_with_dynamic_services():
"""
We need to look at the file, then fetch service names
and preferably service folders
"""
def get_available_commands() -> List[str]: def get_available_commands() -> List[str]:
""" """
Generate all commands with combinatorics. Generate all commands with combinatorics.

View file

@ -1,8 +1,9 @@
from typing import List from typing import List
from time import sleep from time import sleep
import socket as socket_module
# from subprocess import check_output # from subprocess import check_output
from selfprivacy_api.root_daemon import SOCKET_PATH, socket_module from selfprivacy_api.root_daemon import SOCKET_PATH
from tests.test_common import get_test_mode from tests.test_common import get_test_mode

View file

@ -5,7 +5,7 @@ import logging
from typing import List, Optional from typing import List, Optional
from os.path import exists from os.path import exists
from subprocess import run from subprocess import run
from selfprivacy_api.utils.root_interface import call_root_function from selfprivacy_api.root_daemon.root_interface import call_root_function
from selfprivacy_api import utils from selfprivacy_api import utils
from selfprivacy_api.services.config_item import ServiceConfigItem from selfprivacy_api.services.config_item import ServiceConfigItem

View file

@ -1,23 +1,23 @@
import pytest import pytest
import os
import asyncio
import threading
import subprocess import subprocess
from typing import List
from time import sleep
from selfprivacy_api.root_daemon import ( from os import path
from os.path import join, exists
from os import chdir
import selfprivacy_api
import tests
from selfprivacy_api.root_daemon.daemon import (
get_available_commands, get_available_commands,
init, init,
main,
service_commands, service_commands,
services, services,
) )
import selfprivacy_api import selfprivacy_api.root_daemon.daemon as root_daemon
import selfprivacy_api.root_daemon as root_daemon from selfprivacy_api.root_daemon.root_interface import call_root_function
from selfprivacy_api.utils.root_interface import call_root_function
from os.path import join, exists
from typing import List
from time import sleep
@pytest.fixture() @pytest.fixture()
@ -48,16 +48,35 @@ def test_init():
assert sock is not None assert sock is not None
def get_root_sp_directory() -> str:
package_file = selfprivacy_api.__file__
tests_file = tests.__file__
sp_dir = path.commonpath([package_file, tests_file])
# raise ValueError(sp_dir,package_file, tests_file)
return sp_dir
def start_root_demon(): def start_root_demon():
root_daemon_file = selfprivacy_api.root_daemon.__file__
old_dir = path.abspath(path.curdir)
chdir(get_root_sp_directory())
assert path.abspath(path.curdir) == get_root_sp_directory()
# this is a prototype of how we need to run it` # this is a prototype of how we need to run it`
proc = subprocess.Popen(args=["python", root_daemon_file], shell=False) proc = subprocess.Popen(
args=["python", "-m", "selfprivacy_api.root_daemon.daemon"], shell=False
)
# check that it did not error out # check that it did not error out
sleep(0.3) sleep(0.3)
finished = proc.poll() finished = proc.poll()
assert finished is None assert finished is None
chdir(old_dir)
return proc return proc