Merge pull request 'import code' (#2) from dev into master

Reviewed-on: https://chacha.ddns.net/gitea/chacha/dabdatasync/pulls/2
new-tag:0.1.0
This commit was merged in pull request #2.
This commit is contained in:
2024-03-23 13:00:47 +01:00
4 changed files with 99 additions and 53 deletions

View File

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

View File

@@ -0,0 +1,92 @@
import json
from abc import ABC, abstractmethod
from typing import final
from typing import Self, Any, Set
class I_DataSync(ABC):
manifest_path: str = "/opt/pyDABFactoryAppliance/Manifest.json"
@classmethod
@final
def get_manifest_data(cls) -> dict[Any, Any]:
with open(cls.manifest_path) as f_DAB_manifest:
return json.load(f_DAB_manifest)
@classmethod
@final
def try_get_instance(cls, manifest) -> Self | None:
if cls.test_applicable(manifest):
return cls(manifest)
return None
def __init__(self, manifest: dict[Any, Any]) -> None:
self.manifest: dict[Any, Any] = manifest
@classmethod
def test_applicable(cls, manifest: dict[Any, Any]) -> bool:
return False
@abstractmethod
def configure(self) -> None:
pass
@abstractmethod
def connect(self) -> None:
pass
@abstractmethod
def read_data(self) -> None:
pass
@abstractmethod
def write_data(self) -> None:
pass
class DataSync_Factory:
ar_cls_DataSync: Set[I_DataSync] = set()
@classmethod
def register_C_DataSync(cls, cls_DataSync):
cls.ar_cls_DataSync.add(cls_DataSync)
@classmethod
def get_DataSync(cls) -> I_DataSync | None:
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):
cls.register_C_DataSync(_cls)
return _cls
@DataSync_Factory.register
class C_DataSync_NextCloud(I_DataSync):
@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
return False
else:
raise RuntimeError("Wrong manifest format")
return True
def configure(self):
print("Nextcloud configure")
def connect(self):
pass
def read_data(self):
pass
def write_data(self):
pass

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

@@ -8,8 +8,8 @@
import unittest
from os import chdir
from io import StringIO
from contextlib import redirect_stdout,redirect_stderr
from io import StringIO
from contextlib import redirect_stdout, redirect_stderr
from pathlib import Path
print(__name__)
@@ -20,16 +20,13 @@ 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__,"?.?.?")
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 !")
pass