Merge pull request 'dev' (#3) from dev into master
Reviewed-on: https://chacha.ddns.net/gitea/chacha/dabdatasync/pulls/3 new-tag:0.2.0
This commit was merged in pull request #3.
This commit is contained in:
@@ -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/"
|
||||
|
||||
@@ -11,4 +11,5 @@ Main module __init__ file.
|
||||
"""
|
||||
|
||||
from .__metadata__ import __version__, __Summuary__, __Name__
|
||||
from .datasync import I_DataSync, DataSync_Factory
|
||||
from .datasync import I_DataSync, DataSync_Factory, C_DataSync_NextCloud
|
||||
from .datasync import DataSyncException, DataSyncException_InvalidManifest
|
||||
|
||||
@@ -1,58 +1,240 @@
|
||||
"""Main datasync class"""
|
||||
|
||||
import json
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import final
|
||||
from typing import Self, Any, Set
|
||||
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, TemporaryDirectory
|
||||
import shutil
|
||||
|
||||
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"""
|
||||
|
||||
suffix: str
|
||||
|
||||
@abstractmethod
|
||||
def compress(self, path_in: Path, file_out: IO):
|
||||
"""compress method - virtual"""
|
||||
|
||||
@abstractmethod
|
||||
def decompress(self, path_in: Path, path_out: Path):
|
||||
"""decompress method - virtual"""
|
||||
|
||||
|
||||
class DataSync_Compressor_targz(A_DataSync_Compressor):
|
||||
"""Concrete compressor class - .tar.gz compressor"""
|
||||
|
||||
suffix: str = ".tar.gz"
|
||||
|
||||
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))
|
||||
|
||||
def decompress(self, path_in: Path, path_out: Path):
|
||||
"""decompress method - .tar.gz concrete"""
|
||||
with tarfile.open(path_in, "r") as tar:
|
||||
tar.extractall(path_out)
|
||||
|
||||
|
||||
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"""
|
||||
|
||||
@abstractmethod
|
||||
def decompress(self, compressor: A_DataSync_Compressor, path_in: Path) -> 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)
|
||||
|
||||
def decompress(self, compressor: A_DataSync_Compressor, path_in: Path) -> None:
|
||||
"""compress the record - concrete FS implementation"""
|
||||
if TYPE_CHECKING:
|
||||
assert isinstance(self.path, Path)
|
||||
if os.path.isdir(self.path):
|
||||
shutil.rmtree(self.path)
|
||||
if os.path.isfile(self.path):
|
||||
os.remove(self.path)
|
||||
compressor.decompress(path_in, self.path.parent)
|
||||
|
||||
|
||||
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]:
|
||||
with open(cls.manifest_path) as f_DAB_manifest:
|
||||
def get_manifest_data(cls) -> dict[str, 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:
|
||||
def try_get_instance(cls, manifest: dict[str, Any]) -> 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.manifest: dict[Any, Any] = manifest
|
||||
def __init__(self, manifest: dict[str, Any]) -> None:
|
||||
self.connected: bool = False
|
||||
self.compressor: A_DataSync_Compressor = type(self).cls_compressor()
|
||||
self.manifest: dict[str, 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:
|
||||
def test_applicable(cls, manifest: dict[str, Any]) -> bool:
|
||||
"""check if a concrete class is applicable - generic"""
|
||||
del manifest # quality warning removal
|
||||
return False
|
||||
|
||||
@abstractmethod
|
||||
def configure(self) -> None:
|
||||
pass
|
||||
"""configure the class instance"""
|
||||
self._impl_configure()
|
||||
|
||||
@abstractmethod
|
||||
def _impl_configure(self) -> None:
|
||||
"""configure the class instance - virtual"""
|
||||
|
||||
def connect(self) -> None:
|
||||
pass
|
||||
"""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:
|
||||
pass
|
||||
"""read data from the service"""
|
||||
self.connect()
|
||||
with TemporaryDirectory() as tmpdir:
|
||||
for datasync_record in self.ar_datasync_record:
|
||||
self._impl_read_data(Path(datasync_record.name + self.compressor.suffix), Path(tmpdir))
|
||||
datasync_record.decompress(self.compressor, Path(tmpdir) / (datasync_record.name + self.compressor.suffix))
|
||||
|
||||
@abstractmethod
|
||||
def _impl_read_data(self, file_in: Path, file_out: Path) -> None:
|
||||
"""read data from the service - virtual"""
|
||||
|
||||
def write_data(self) -> None:
|
||||
pass
|
||||
"""write data to the service"""
|
||||
self.connect()
|
||||
self._impl_wipe_data()
|
||||
for datasync_record in self.ar_datasync_record:
|
||||
try:
|
||||
with NamedTemporaryFile("wb", suffix=self.compressor.suffix, 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"""
|
||||
|
||||
def wipe_data(self) -> None:
|
||||
"""wipe data on the service"""
|
||||
self.connect()
|
||||
self._impl_wipe_data()
|
||||
|
||||
@abstractmethod
|
||||
def _impl_wipe_data(self) -> None:
|
||||
"""wipe data on the service - virtual"""
|
||||
|
||||
|
||||
class DataSync_Factory:
|
||||
ar_cls_DataSync: Set[I_DataSync] = set()
|
||||
"""DataSync Factory"""
|
||||
|
||||
@classmethod
|
||||
def register_C_DataSync(cls, cls_DataSync):
|
||||
cls.ar_cls_DataSync.add(cls_DataSync)
|
||||
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):
|
||||
@@ -61,32 +243,88 @@ class DataSync_Factory:
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def register(cls, _cls):
|
||||
cls.register_C_DataSync(_cls)
|
||||
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 "NextCloudSync" in manifest["Args"]:
|
||||
if manifest["Args"]["NextCloudSync"]["value"] is True:
|
||||
print("Nextcloud Sync found")
|
||||
return True
|
||||
def test_applicable(cls, manifest: dict[str, Any]) -> bool:
|
||||
"""check if a concrete class is applicable - Nextcloud override"""
|
||||
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 - Nextcloud concrete implementation"""
|
||||
if "FSSync_NextCloud_Address" in self.manifest["Args"]:
|
||||
self.nextcloud_address = self.manifest["Args"]["FSSync_NextCloud_Address"]["value"]
|
||||
else:
|
||||
raise RuntimeError("Wrong manifest format")
|
||||
return True
|
||||
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, "/"
|
||||
)
|
||||
|
||||
def configure(self):
|
||||
print("Nextcloud configure")
|
||||
else:
|
||||
raise DataSyncException_InvalidManifest()
|
||||
|
||||
def connect(self):
|
||||
pass
|
||||
def _impl_connect(self) -> None:
|
||||
"""connect to the remote service - Nextcloud 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 read_data(self):
|
||||
pass
|
||||
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 write_data(self):
|
||||
pass
|
||||
def _impl_read_data(self, file_in: Path, file_out: Path) -> None:
|
||||
"""read data from the remote service - Nextcloud concrete implementation"""
|
||||
self._check_create_dir()
|
||||
self.client.download_sync(
|
||||
remote_path=str(self.nextcloud_path / file_in).replace(os.sep, "/"), local_path=str(file_out / file_in).replace(os.sep, "/")
|
||||
)
|
||||
|
||||
def _impl_write_data(self, record_name: str, file_in: IO) -> None:
|
||||
"""write data to the remote service - Nextcloud 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,
|
||||
)
|
||||
|
||||
def _impl_wipe_data(self) -> None:
|
||||
"""wipe data on the service - concrete implementation"""
|
||||
if self.client.check(self.nextcloud_path):
|
||||
self.client.clean(self.nextcloud_path)
|
||||
|
||||
1
test/test_data_origin/SAVE_FILE.txt
Normal file
1
test/test_data_origin/SAVE_FILE.txt
Normal file
@@ -0,0 +1 @@
|
||||
SAVED_VALUE
|
||||
@@ -8,9 +8,9 @@
|
||||
|
||||
import unittest
|
||||
from os import chdir
|
||||
from io import StringIO
|
||||
from contextlib import redirect_stdout, redirect_stderr
|
||||
from pathlib import Path
|
||||
import pprint
|
||||
import shutil
|
||||
|
||||
print(__name__)
|
||||
print(__package__)
|
||||
@@ -21,12 +21,109 @@ testdir_path = Path(__file__).parent.resolve()
|
||||
chdir(testdir_path.parent.resolve())
|
||||
|
||||
|
||||
class Testtest_module(unittest.TestCase):
|
||||
class TestDabDataSync(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
chdir(testdir_path.parent.resolve())
|
||||
shutil.rmtree(testdir_path / "test_data", ignore_errors=True)
|
||||
shutil.rmtree(testdir_path / "test_data2", ignore_errors=True)
|
||||
shutil.copytree(testdir_path / "test_data_origin", testdir_path / "test_data")
|
||||
shutil.copytree(testdir_path / "test_data_origin", testdir_path / "test_data2")
|
||||
|
||||
def tearDown(self) -> None:
|
||||
shutil.rmtree(testdir_path / "test_data", ignore_errors=True)
|
||||
shutil.rmtree(testdir_path / "test_data2", ignore_errors=True)
|
||||
|
||||
def test_version(self):
|
||||
self.assertNotEqual(dabdatasync.__version__, "?.?.?")
|
||||
|
||||
def test_test_module(self):
|
||||
pass
|
||||
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)
|
||||
|
||||
self.assertEqual(len(datasync.ar_datasync_record), 2)
|
||||
self.assertEqual(datasync.ar_datasync_record[0].name, "SOTF_map")
|
||||
self.assertEqual(datasync.ar_datasync_record[1].name, "SOTF_map2")
|
||||
self.assertEqual(datasync.ar_datasync_record[0].rec_type, "fs")
|
||||
self.assertEqual(datasync.ar_datasync_record[1].rec_type, "fs")
|
||||
self.assertEqual(datasync.ar_datasync_record[0].value, "test/test_data")
|
||||
self.assertEqual(datasync.ar_datasync_record[1].value, "test/test_data2/SAVE_FILE.txt")
|
||||
|
||||
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
||||
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
||||
|
||||
datasync.write_data()
|
||||
|
||||
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
||||
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
||||
|
||||
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
||||
testfile.write("MODIFIED_VALUE")
|
||||
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
||||
testfile.write("MODIFIED_VALUE2")
|
||||
|
||||
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "MODIFIED_VALUE")
|
||||
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "MODIFIED_VALUE2")
|
||||
|
||||
datasync.read_data()
|
||||
|
||||
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
||||
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
||||
|
||||
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
||||
testfile.write("MODIFIED_VALUE3")
|
||||
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
||||
testfile.write("MODIFIED_VALUE32")
|
||||
|
||||
datasync.write_data()
|
||||
|
||||
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "MODIFIED_VALUE3")
|
||||
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "MODIFIED_VALUE32")
|
||||
|
||||
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
||||
testfile.write("MODIFIED_VALUE")
|
||||
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
||||
testfile.write("MODIFIED_VALUE")
|
||||
|
||||
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "MODIFIED_VALUE")
|
||||
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "MODIFIED_VALUE")
|
||||
|
||||
datasync.read_data()
|
||||
|
||||
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "MODIFIED_VALUE3")
|
||||
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
||||
self.assertEqual(testfile.read(), "MODIFIED_VALUE32")
|
||||
|
||||
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()
|
||||
|
||||
3
test/test_manifest_empty.json
Normal file
3
test/test_manifest_empty.json
Normal file
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"APP_ID": "2a13dff2-1298-11ee-be56-0242ac120002",
|
||||
"Args": {}}
|
||||
2
test/test_manifest_invalid.json
Normal file
2
test/test_manifest_invalid.json
Normal file
@@ -0,0 +1,2 @@
|
||||
{
|
||||
"APP_ID": "2a13dff2-1298-11ee-be56-0242ac120002"}
|
||||
541
test/test_manifest_nextcloud.json
Normal file
541
test/test_manifest_nextcloud.json
Normal file
@@ -0,0 +1,541 @@
|
||||
{
|
||||
"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"
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"type": "T_FSSYNC_RECORD",
|
||||
"value": {
|
||||
"name": {
|
||||
"type": "SIMPLE_STRING",
|
||||
"value": "SOTF_map2"
|
||||
},
|
||||
"type": {
|
||||
"type": "SIMPLE_STRING",
|
||||
"value": "fs"
|
||||
},
|
||||
"value": {
|
||||
"type": "STRING",
|
||||
"value": "test/test_data2/SAVE_FILE.txt"
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
"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": []
|
||||
}
|
||||
}
|
||||
}
|
||||
9
test/test_manifest_nextcloud_disabled.json
Normal file
9
test/test_manifest_nextcloud_disabled.json
Normal 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"}
|
||||
}}
|
||||
5
test/test_manifest_nextcloud_invalid.json
Normal file
5
test/test_manifest_nextcloud_invalid.json
Normal file
@@ -0,0 +1,5 @@
|
||||
{
|
||||
"APP_ID": "2a13dff2-1298-11ee-be56-0242ac120002",
|
||||
"Args": {
|
||||
"FSSync_NextCloud_Enabled": {"type": "BOOL", "value": true}
|
||||
}}
|
||||
Reference in New Issue
Block a user