5 Commits

Author SHA1 Message Date
cclecle
38031deba2 fix: ignore webdav3.client typing
fix: ignore coverage for 'if TYPE_CHECKING:' directives
2024-03-28 00:02:18 +00:00
cclecle
abe800814d fix quality and typing 2024-03-27 02:37:47 +00:00
cclecle
81ba24d6f3 improve quality 2024-03-27 02:17:08 +00:00
cclecle
b56a61b796 implement Nextcloud write 2024-03-27 01:52:19 +00:00
cclecle
f1a748c09f import code 2024-03-23 11:56:52 +00:00
12 changed files with 896 additions and 83 deletions

View File

@@ -34,7 +34,9 @@ classifiers = [
]
dependencies = [
'importlib-metadata; python_version<"3.9"',
'packaging'
'packaging',
'webdavclient3',
'pydantic'
]
dynamic = ["version"]
@@ -49,9 +51,9 @@ where = ["src"]
"dabdatasync.data" = ["*.*"]
"dabdatasync" = ["py.typed"]
# [[tool.mypy.overrides]]
# module = ""
# ignore_missing_imports = true
[[tool.mypy.overrides]]
module = "webdav3.client"
ignore_missing_imports = true
[tool.coverage.run]
cover_pylib = false
@@ -63,6 +65,11 @@ concurrency = [
'thread'
]
[tool.coverage.report]
exclude_also = [
"if TYPE_CHECKING:",
]
[project.urls]
Homepage = "https://chacha.ddns.net/gitea/chacha/dabdatasync"
Documentation = "https://chacha.ddns.net/mkdocs-web/chacha/dabdatasync/master/latest/"

View File

@@ -11,4 +11,5 @@ Main module __init__ file.
"""
from .__metadata__ import __version__, __Summuary__, __Name__
from .test_module import test_function
from .datasync import I_DataSync, DataSync_Factory, C_DataSync_NextCloud
from .datasync import DataSyncException, DataSyncException_InvalidManifest

280
src/dabdatasync/datasync.py Normal file
View File

@@ -0,0 +1,280 @@
"""Main datasync class"""
import json
from abc import ABC, abstractmethod
from typing import final, TYPE_CHECKING, IO
from typing import Self, Any, Set, Optional
from uuid import UUID
from pathlib import Path
import os
import tarfile
from tempfile import NamedTemporaryFile
from pydantic import BaseModel
from webdav3.client import Client as webdav3_Client
class DataSyncException(Exception):
"""generic datasync exception class"""
class DataSyncException_InvalidManifest(DataSyncException):
"""specific datasync exception class - Dab appliance manifest not found"""
def urljoin(*args):
"""
Joins given arguments into an url. Trailing but not leading slashes are
stripped for each argument.
"""
return "/".join(map(lambda x: str(x).rstrip("/"), args))
class A_DataSync_Compressor(ABC):
"""abstract compressor class"""
@abstractmethod
def compress(self, path_in: Path, file_out: IO):
"""compress method - virtual"""
class DataSync_Compressor_targz(A_DataSync_Compressor):
"""Concrete compressor class - .tar.gz compressor"""
def compress(self, path_in: Path, file_out: IO):
"""compress method - .tar.gz concrete"""
with tarfile.open(fileobj=file_out, mode="w:gz") as tar:
tar.add(path_in, arcname=os.path.basename(path_in))
class A_DataSync_Record(BaseModel, ABC):
"""Abstract DataSync Record class"""
name: str
rec_type: str
value: str
@abstractmethod
def compress(self, compressor: A_DataSync_Compressor, file_out: IO) -> None:
"""compress the record - virtual"""
class DataSync_Record_Factory:
"""DataSync Record Factory"""
ar_cls_DataSync_Record: Set[type[A_DataSync_Record]] = set()
@classmethod
def get_C_DataSync_Record(cls, name: str, rec_type: str, value: str) -> A_DataSync_Record | None:
"""get a concrete DataSync Record class instance"""
for cls_DataSync_Record in cls.ar_cls_DataSync_Record:
if cls_DataSync_Record.model_fields["rec_type"].default == rec_type:
return C_DataSync_Record_FS(name=name, rec_type=rec_type, value=value)
raise RuntimeError("No DataSync_Record concrete class found")
@classmethod
def register(cls, _cls: type[A_DataSync_Record]) -> type[A_DataSync_Record]:
"""decorator to register a concrete DataSync Record class"""
cls.ar_cls_DataSync_Record.add(_cls)
return _cls
@DataSync_Record_Factory.register
class C_DataSync_Record_FS(A_DataSync_Record):
"""Concrete DataSync Record class - FileSystem"""
rec_type: str = "fs"
path: Optional[Path] = None
def model_post_init(self, __context) -> None:
self.path = Path(self.value)
def compress(self, compressor: A_DataSync_Compressor, file_out: IO) -> None:
"""compress the DataSync Record - concrete FS implementation"""
if TYPE_CHECKING:
assert isinstance(self.path, Path)
compressor.compress(self.path, file_out)
class I_DataSync(ABC):
"""Abstract DataSync class"""
manifest_path: str = "/opt/pyDABFactoryAppliance/Manifest.json"
cls_compressor: type[A_DataSync_Compressor] = DataSync_Compressor_targz
@classmethod
@final
def get_manifest_data(cls) -> dict[Any, Any]:
"""utilitary method to get manifest"""
with open(cls.manifest_path, encoding="utf-8") as f_DAB_manifest:
return json.load(f_DAB_manifest)
@classmethod
@final
def try_get_instance(cls, manifest) -> Self | None:
"""try to get an instance of a concrete class"""
if cls.test_applicable(manifest):
return cls(manifest)
return None
def __init__(self, manifest: dict[Any, Any]) -> None:
self.connected: bool = False
self.compressor: A_DataSync_Compressor = type(self).cls_compressor()
self.manifest: dict[Any, Any] = manifest
self.app_id: UUID = UUID(manifest["APP_ID"])
self.ar_datasync_record: list[A_DataSync_Record] = []
if "FSSYNC_RECORD" in manifest["Args"]:
for record in manifest["Args"]["FSSYNC_RECORD"]["value"]:
record = DataSync_Record_Factory.get_C_DataSync_Record(
record["value"]["name"]["value"],
record["value"]["type"]["value"],
record["value"]["value"]["value"],
)
assert isinstance(record, A_DataSync_Record)
self.ar_datasync_record.append(record)
@classmethod
def test_applicable(cls, manifest: dict[Any, Any]) -> bool:
"""check if a concrete class is applicable"""
del manifest # quality warning removal
return False
def configure(self) -> None:
"""configure the class instance"""
self._impl_configure()
@abstractmethod
def _impl_configure(self) -> None:
"""configure the class instance - virtual"""
def connect(self) -> None:
"""connect to the service"""
if not self.connected:
self._impl_connect()
self.connected = True
@abstractmethod
def _impl_connect(self) -> None:
"""connect to the service - virtual"""
def read_data(self) -> None:
"""read data from the service"""
self.connect()
self._impl_read_data()
@abstractmethod
def _impl_read_data(self) -> None:
"""read data from the service - virtual"""
def write_data(self) -> None:
"""write data to the service"""
self.connect()
for datasync_record in self.ar_datasync_record:
try:
with NamedTemporaryFile("wb", suffix=".tar.gz", delete=False) as tmp_file:
datasync_record.compress(self.compressor, tmp_file)
tmp_file.seek(0)
tmp_file.close()
self._impl_write_data(datasync_record.name + "".join(Path(tmp_file.name).suffixes), tmp_file)
finally:
os.unlink(tmp_file.name)
@abstractmethod
def _impl_write_data(self, record_name: str, file_in: IO) -> None:
"""write data to the service - virtual"""
class DataSync_Factory:
"""DataSync Factory"""
ar_cls_DataSync: Set[type[I_DataSync]] = set()
@classmethod
def get_DataSync(cls) -> I_DataSync | None:
"""get and configure a DataSync Concrete class instance"""
manifest = I_DataSync.get_manifest_data()
for cls_DataSync in cls.ar_cls_DataSync:
if res := cls_DataSync.try_get_instance(manifest):
res.configure()
return res
return None
@classmethod
def register(cls, _cls: type[I_DataSync]) -> type[I_DataSync]:
"""decorator to register a concrete class to the factory"""
cls.ar_cls_DataSync.add(_cls)
return _cls
@DataSync_Factory.register
class C_DataSync_NextCloud(I_DataSync):
"""Concrete DataSync class - Nextcloud"""
def __init__(self, manifest: dict[Any, Any]) -> None:
super().__init__(manifest)
self.nextcloud_address: str
self.nextcloud_user: str
self.nextcloud_password: str
self.nextcloud_path: str
self.client: webdav3_Client
self.connected: bool = False
@classmethod
def test_applicable(cls, manifest) -> bool:
if "Args" in manifest:
if "FSSync_NextCloud_Enabled" in manifest["Args"]:
if manifest["Args"]["FSSync_NextCloud_Enabled"]["value"] is True:
return True
return False
raise DataSyncException_InvalidManifest()
def _impl_configure(self) -> None:
"""configure the class instance - concrete implementation"""
if "FSSync_NextCloud_Address" in self.manifest["Args"]:
self.nextcloud_address = self.manifest["Args"]["FSSync_NextCloud_Address"]["value"]
else:
raise DataSyncException_InvalidManifest()
if "FSSync_NextCloud_User" in self.manifest["Args"]:
self.nextcloud_user = self.manifest["Args"]["FSSync_NextCloud_User"]["value"]
else:
raise DataSyncException_InvalidManifest()
if "FSSync_NextCloud_Password" in self.manifest["Args"]:
self.nextcloud_password = self.manifest["Args"]["FSSync_NextCloud_Password"]["value"]
else:
raise DataSyncException_InvalidManifest()
if "FSSync_NextCloud_Path" in self.manifest["Args"]:
self.nextcloud_path = str(Path(self.manifest["Args"]["FSSync_NextCloud_Path"]["value"]) / Path(str(self.app_id))).replace(
os.sep, "/"
)
else:
raise DataSyncException_InvalidManifest()
def _impl_connect(self) -> None:
"""connect to the remote service - concrete implementation"""
full_adress = urljoin(self.nextcloud_address, "remote.php/dav/files/", self.nextcloud_user)
self.client = webdav3_Client(
{"webdav_hostname": full_adress, "webdav_login": self.nextcloud_user, "webdav_password": self.nextcloud_password}
)
def _check_create_dir(self) -> None:
"""check and create directory in remote service"""
url_accumulator: str = ""
for url_part in self.nextcloud_path.split("/"):
url_accumulator += "/" + url_part
if not self.client.check(url_accumulator):
self.client.mkdir(url_accumulator)
def _impl_read_data(self) -> None:
"""read data from the remote service - concrete implementation"""
self._check_create_dir()
def _impl_write_data(self, record_name: str, file_in: IO) -> None:
"""write data to the remote service - concrete implementation"""
self._check_create_dir()
self.client.upload_sync(
remote_path=str(Path(self.nextcloud_path) / record_name).replace(os.sep, "/"),
local_path=file_in.name,
)

View File

@@ -1,43 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# dabdatasync (c) by chacha
#
# dabdatasync is licensed under a
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
#
# You should have received a copy of the license along with this
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
"""Phasellus tellus lectus, volutpat eu dapibus ut, suscipit vel augue.
Tips:
Aliquam non leo vel libero sagittis viverra. Quisque lobortis nunc sit amet augue euismod laoreet.
Note:
Maecenas volutpat porttitor pretium. Aliquam suscipit quis nisi non imperdiet.
Note:
Vivamus et efficitur lorem, eget imperdiet tortor. Integer vel interdum sem.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING: # Only imports the below statements during type checking
pass
def test_function(testvar: int) -> int:
""" A test function that return testvar+1 and print "Hello world !"
Proin eget sapien eget ipsum efficitur mollis nec ac nibh.
Note:
Morbi id lectus maximus, condimentum nunc eget, porta felis. In tristique velit tortor.
Args:
testvar: any integer
Returns:
testvar+1
"""
print("Hello world !")
return testvar+1

View File

@@ -0,0 +1 @@
SAVED_VALUE

59
test/test_datasync.py Normal file
View File

@@ -0,0 +1,59 @@
# dabdatasync (c) by chacha
#
# dabdatasync is licensed under a
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
#
# You should have received a copy of the license along with this
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
import unittest
from os import chdir
from io import StringIO
from contextlib import redirect_stdout, redirect_stderr
from pathlib import Path
import pprint
print(__name__)
print(__package__)
from src import dabdatasync
testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())
class TestDabDataSync(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
def test_version(self):
self.assertNotEqual(dabdatasync.__version__, "?.?.?")
def test_load_nextcloud(self):
dabdatasync.I_DataSync.manifest_path = testdir_path / "test_manifest_nextcloud.json"
datasync = dabdatasync.DataSync_Factory.get_DataSync()
self.assertIsInstance(datasync, dabdatasync.I_DataSync)
self.assertIsInstance(datasync, dabdatasync.C_DataSync_NextCloud)
datasync.read_data()
pprint.pprint(datasync.ar_datasync_record)
datasync.write_data()
def test_load_empty(self):
dabdatasync.I_DataSync.manifest_path = testdir_path / "test_manifest_empty.json"
datasync = dabdatasync.DataSync_Factory.get_DataSync()
self.assertIsNone(datasync)
def test_load_nextcloud_disabled(self):
dabdatasync.I_DataSync.manifest_path = testdir_path / "test_manifest_nextcloud_disabled.json"
datasync = dabdatasync.DataSync_Factory.get_DataSync()
self.assertIsNone(datasync)
def test_load_invalid(self):
dabdatasync.I_DataSync.manifest_path = testdir_path / "test_manifest_invalid.json"
with self.assertRaises(dabdatasync.DataSyncException_InvalidManifest):
dabdatasync.DataSync_Factory.get_DataSync()
def test_load_nextcloud_invalid(self):
dabdatasync.I_DataSync.manifest_path = testdir_path / "test_manifest_nextcloud_invalid.json"
with self.assertRaises(dabdatasync.DataSyncException_InvalidManifest):
dabdatasync.DataSync_Factory.get_DataSync()

View File

@@ -0,0 +1,3 @@
{
"APP_ID": "2a13dff2-1298-11ee-be56-0242ac120002",
"Args": {}}

View File

@@ -0,0 +1,2 @@
{
"APP_ID": "2a13dff2-1298-11ee-be56-0242ac120002"}

View File

@@ -0,0 +1,524 @@
{
"APP_NAME": "CHACHA-SOTF",
"APP_DESC": "ChaCha SonOfTheForset Dedicated Server",
"APP_ID": "2a13dff2-1298-11ee-be56-0242ac120002",
"REFERENCE_CONFIG_ID": "cf698a62-120a-11ee-be56-0242ac120002",
"VIRTUAL": false,
"NOBOOTSTRAP": false,
"NOFINALIZE": false,
"NOSTART": false,
"creation_date": "2024-03-24T19:07:48.862542",
"Params": {
"ROOTFS_SIZE_G": {
"value": 20,
"modified": true
},
"AR_TAGS": {
"value": [
{
"value": "pydabfactory"
},
{
"value": "debianbase"
},
{
"value": "pydabfactory"
},
{
"value": "chacha"
},
{
"value": "pydabfactory"
},
{
"value": "games"
},
{
"value": "pydabfactory"
},
{
"value": "sotf"
}
],
"modified": true
},
"FEATURE_NESTING": {
"value": false,
"modified": false
},
"MAIN_MACADDR": {
"value": "D2:A9:59:72:C4:B4",
"modified": true
},
"AUTOSTART": {
"value": true,
"modified": true
},
"CPU_UNIT": {
"value": 1536,
"modified": true
},
"PRIVILEGIED": {
"value": false,
"modified": false
},
"DEST_NODE": {
"value": "hypervisor2",
"modified": true
},
"SWAP_M": {
"value": 2048,
"modified": true
},
"AR_CFG_OPT": {
"value": [],
"modified": false
},
"RUNNING_STORAGE": {
"value": "VMStore2",
"modified": true
},
"FEATURE_FUSE": {
"value": false,
"modified": false
},
"FEATURE_MKNODE": {
"value": false,
"modified": false
},
"NETWORK_BRIDGE": {
"value": "vmbr1",
"modified": true
},
"CPU_COUNT": {
"value": 2,
"modified": true
},
"TEMPLATE_STORAGE": {
"value": "live-storage-h2",
"modified": true
},
"RAM_M": {
"value": 12000,
"modified": true
}
},
"Args": {
"RootPasswd": {
"type": "ROOT_PASSWD",
"value": "######"
},
"DEBUG_TOOLS": {
"type": "BOOL",
"value": false
},
"SSH_PORT": {
"type": "LISTEN_PORT",
"value": {
"port_type": {
"type": "STRING",
"value": "tcp"
},
"value": {
"type": "UINT",
"value": 50020
}
}
},
"FirstBootFilePath": {
"type": "STRING",
"value": "/firstboot.sh"
},
"CustomFirstBootFilePath": {
"type": "STRING",
"value": "/customfirstboot.sh"
},
"ACLSavePath": {
"type": "STRING",
"value": "/saved.acl"
},
"locale": {
"type": "STRING",
"value": "fr_FR.UTF-8"
},
"locale_gen": {
"type": "STRING",
"value": "fr_FR.UTF-8 UTF-8"
},
"timezone": {
"type": "STRING",
"value": "Europe/Paris"
},
"EnableLog2Ram": {
"type": "BOOL",
"value": false
},
"EnableAutoReboot": {
"type": "BOOL",
"value": true
},
"EnableJava": {
"type": "BOOL",
"value": false
},
"ForcePython39": {
"type": "BOOL",
"value": false
},
"EXEC_USER": {
"type": "STRING",
"value": "GenUser"
},
"EXEC_USER_ID": {
"type": "UINT",
"value": 1000
},
"EXEC_USER_PASSWD": {
"type": "PASSWD",
"value": "######"
},
"DEFAULT_CHACHA_GIT_BRANCH": {
"type": "STRING",
"value": "production"
},
"SECTION": {
"type": "STRING",
"value": "games"
},
"SystemDJournalMaxSize": {
"type": "STRING",
"value": "40M"
},
"FSSYNC_PRESYNC_CMD": {
"type": "STRING",
"value": ""
},
"FSSYNC_POSTSYNC_CMD": {
"type": "STRING",
"value": ""
},
"FSSYNC_INITIAL_FETCH": {
"type": "BOOL",
"value": true
},
"FSSYNC_RECORD": {
"type": "T_ARRAY_FSSYNC_RECORD",
"value": [
{
"type": "T_FSSYNC_RECORD",
"value": {
"name": {
"type": "SIMPLE_STRING",
"value": "SOTF_map"
},
"type": {
"type": "SIMPLE_STRING",
"value": "fs"
},
"value": {
"type": "STRING",
"value": "test/test_data"
}
}
}
]
},
"FSSync_NextCloud_Enabled": {
"type": "BOOL",
"value": true
},
"FSSync_NextCloud_Address": {
"type": "URL",
"value": "https://chacha.ddns.net/nextcloud"
},
"FSSync_NextCloud_User": {
"type": "STRING",
"value": "chacha-bot"
},
"FSSync_NextCloud_Password": {
"type": "STRING",
"value": "F3P8m-nQHik-NSmb2-mnFEF-s85RE"
},
"FSSync_NextCloud_Path": {
"type": "STRING",
"value": "pydabfactory-test"
},
"GAME_MNG_PWD": {
"type": "STRING",
"value": "cfographut"
},
"GAME_MNG_DEFAULT_MODE": {
"type": "STRING",
"value": "BLACKLIST"
},
"GAME_MNG_LISTENING_PORT": {
"type": "UINT",
"value": 50000
},
"GAME_MNG_RESTART_DELAY": {
"type": "UINT",
"value": 30
},
"GAMETYPENAME": {
"type": "STRING",
"value": "sotf"
},
"WINE_ARCH": {
"type": "STRING",
"value": "win64"
},
"WINE_NAME": {
"type": "STRING",
"value": "wine-9.4-staging-tkg-amd64"
},
"WINEPREFIX": {
"type": "STRING",
"value": "DEFAULT"
},
"WINE_URL": {
"type": "URL",
"value": "https://github.com/Kron4ek/Wine-Builds/releases/download/9.4/wine-9.4-staging-tkg-amd64.tar.xz"
},
"ENABLE_WINE_ESYNC": {
"type": "BOOL",
"value": false
},
"ENABLE_WINE_FSYNC": {
"type": "BOOL",
"value": true
},
"ENABLE_WINE_PRELOADER": {
"type": "BOOL",
"value": false
},
"MEMLIMITHIGH": {
"type": "SYSTEMD_RAM",
"value": "10G"
},
"MEMLIMITMAX": {
"type": "SYSTEMD_RAM",
"value": "11G"
},
"CPUQUOTA": {
"type": "SYSTEMD_CPUQUOTA",
"value": 98
},
"STEAM_APP_ID": {
"type": "UINT",
"value": 2465200
},
"STEAM_LOGIN": {
"type": "STRING",
"value": "cclecle"
},
"STEAM_PWD": {
"type": "PASSWD",
"value": "######"
},
"HostName": {
"type": "STRING",
"value": "ChaCha - Sons Of The Forest Server"
},
"MaxPlayers": {
"type": "UINT",
"value": 8
},
"GamePassword": {
"type": "STRING",
"value": "!bourges2023"
},
"GamePort": {
"type": "EXT_LISTEN_PORT",
"value": {
"port_type": {
"type": "STRING",
"value": "udp"
},
"value": {
"type": "UINT",
"value": 8766
}
}
},
"QueryPort": {
"type": "EXT_LISTEN_PORT",
"value": {
"port_type": {
"type": "STRING",
"value": "udp"
},
"value": {
"type": "UINT",
"value": 27018
}
}
},
"BlobSyncPort": {
"type": "EXT_LISTEN_PORT",
"value": {
"port_type": {
"type": "STRING",
"value": "udp"
},
"value": {
"type": "UINT",
"value": 9700
}
}
},
"LanOnly": {
"type": "BOOL",
"value": false
},
"SaveSlot": {
"type": "UINT",
"value": 1
},
"SaveInterval": {
"type": "UINT",
"value": 600
},
"TreeRegrowth": {
"type": "BOOL",
"value": true
},
"StructureDamage": {
"type": "BOOL",
"value": false
},
"EnemySpawn": {
"type": "BOOL",
"value": true
},
"SkipNetworkAccessibilityTest": {
"type": "BOOL",
"value": true
},
"EnemyHealth": {
"type": "STRING",
"value": "Normal"
},
"EnemyDamage": {
"type": "STRING",
"value": "Normal"
},
"EnemyArmour": {
"type": "STRING",
"value": "Normal"
},
"EnemyAggression": {
"type": "STRING",
"value": "Normal"
},
"AnimalSpawnRate": {
"type": "STRING",
"value": "Normal"
},
"StartingSeason": {
"type": "STRING",
"value": "Summer"
},
"SeasonLength": {
"type": "STRING",
"value": "Default"
},
"DayLength": {
"type": "STRING",
"value": "Default"
},
"PrecipitationFrequency": {
"type": "STRING",
"value": "Default"
},
"ConsumableEffects": {
"type": "STRING",
"value": "Normal"
},
"PlayerStatsDamage": {
"type": "STRING",
"value": "Off"
},
"ColdPenalties": {
"type": "STRING",
"value": "Off"
},
"ReducedFoodInContainers": {
"type": "BOOL",
"value": false
},
"SingleUseContainers": {
"type": "BOOL",
"value": false
},
"DAB_LISTEN_PORTS": {
"type": "AR_LISTEN_PORT",
"value": [
{
"type": "LISTEN_PORT",
"value": {
"port_type": {
"type": "STRING",
"value": "tcp"
},
"value": {
"type": "UINT",
"value": 50020
}
}
}
]
},
"DAB_EXT_LISTEN_PORTS": {
"type": "AR_EXT_LISTEN_PORT",
"value": [
{
"type": "EXT_LISTEN_PORT",
"value": {
"port_type": {
"type": "STRING",
"value": "udp"
},
"value": {
"type": "UINT",
"value": 8766
}
}
},
{
"type": "EXT_LISTEN_PORT",
"value": {
"port_type": {
"type": "STRING",
"value": "udp"
},
"value": {
"type": "UINT",
"value": 27018
}
}
},
{
"type": "EXT_LISTEN_PORT",
"value": {
"port_type": {
"type": "STRING",
"value": "udp"
},
"value": {
"type": "UINT",
"value": 9700
}
}
}
]
},
"DAB_MOUNT_POINTS": {
"type": "AR_MOUNT_POINT",
"value": []
},
"DAB_SHARED_MOUNT_POINTS": {
"type": "AR_MOUNT_POINT",
"value": []
}
}
}

View File

@@ -0,0 +1,9 @@
{
"APP_ID": "2a13dff2-1298-11ee-be56-0242ac120002",
"Args": {
"FSSync_NextCloud_Enabled": {"type": "BOOL", "value": false},
"FSSync_NextCloud_Address": {"type": "URL", "value": "https://chacha.ddns.net/nextcloud"},
"FSSync_NextCloud_User": {"type": "STRING", "value": "chacha-bot"},
"FSSync_NextCloud_Password": {"type": "STRING", "value": "F3P8m-nQHik-NSmb2-mnFEF-s85RE"},
"FSSync_NextCloud_Path": {"type": "STRING", "value": "pydabfactory"}
}}

View File

@@ -0,0 +1,5 @@
{
"APP_ID": "2a13dff2-1298-11ee-be56-0242ac120002",
"Args": {
"FSSync_NextCloud_Enabled": {"type": "BOOL", "value": true}
}}

View File

@@ -1,35 +0,0 @@
# dabdatasync (c) by chacha
#
# dabdatasync is licensed under a
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
#
# You should have received a copy of the license along with this
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
import unittest
from os import chdir
from io import StringIO
from contextlib import redirect_stdout,redirect_stderr
from pathlib import Path
print(__name__)
print(__package__)
from src import dabdatasync
testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())
class Testtest_module(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
def test_version(self):
self.assertNotEqual(dabdatasync.__version__,"?.?.?")
def test_test_module(self):
with redirect_stdout(StringIO()) as capted_stdout, redirect_stderr(StringIO()) as capted_stderr:
self.assertEqual(dabdatasync.test_function(41),42)
self.assertEqual(len(capted_stderr.getvalue()),0)
self.assertEqual(capted_stdout.getvalue().strip(),"Hello world !")