Compare commits
1 Commits
Author | SHA1 | Date | |
---|---|---|---|
cf589155f9 |
@ -1 +0,0 @@
|
||||
tar: .
|
18
README.md
18
README.md
@ -1,6 +1,6 @@
|
||||
# alterator_bindings.backend3
|
||||
# alterator-python-functions
|
||||
This module provides bindings to write alterator backends in `Python`.\
|
||||
Pay attention that the module is `alterator_bindings.backend3`
|
||||
Pay attention that the module file is named `alterator_python_functions` (with underscores, not dashes)
|
||||
|
||||
# Description
|
||||
Alterator backends use a very simple communication protocol and can be written in any language. This library contains a set of functions to simplify development of backend on `Python`.
|
||||
@ -19,15 +19,15 @@ Simple backend on `Python` looks like that:
|
||||
```python
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from alterator_bindings import backend3
|
||||
from alterator_bindings.backend3 import *
|
||||
import alterator_python_functions
|
||||
from alterator_python_functions import *
|
||||
|
||||
# Set True if you want to see debug messages
|
||||
backend3.ALTERATOR_DEBUG = True
|
||||
alterator_python_functions.ALTERATOR_DEBUG = True
|
||||
|
||||
# Dictionary to use for translation (usually just alterator module name)
|
||||
# Will set to running backend filename by default (So may be not what you need)
|
||||
backend3.TEXTDOMAIN = "alterator-test"
|
||||
alterator_python_functions.TEXTDOMAIN = "alterator-test"
|
||||
|
||||
# message object is Python's dict, containing pairs attribute-value from Alterator
|
||||
def on_message(message: dict):
|
||||
@ -42,14 +42,14 @@ message_loop(on_message)
|
||||
# Functionality description
|
||||
|
||||
### Globals (may be set by user)
|
||||
- `backend3.ALTERATOR_DEBUG` (Default is `False`) - Indicates if additional debug output (to stderr) is needed (messages from alterator, etc.). Also with this flag `write_debug` won't be silent (will print messages to stderr)
|
||||
- `backend3.TEXTDOMAIN`(Defaults to `alterator-{running backend filename}`) - Sets dictionary used for translations. If not set manually, warning will be printed
|
||||
- `alterator_python_functions.ALTERATOR_DEBUG` (Default is `False`) - Indicates if additional debug output (to stderr) is needed (messages from alterator, etc.). Also with this flag `write_debug` won't be silent (will print messages to stderr)
|
||||
- `alterator_python_functions.TEXTDOMAIN`(Defaults to `alterator-{running backend filename}`) - Sets dictionary used for translations. If not set manually, warning will be printed
|
||||
|
||||
### Functions
|
||||
#### Common functions:
|
||||
- `message_loop` - Main event loop. Message callback should be specified when calling this function.
|
||||
- `on_message` - Your (!!!) callback function. All incoming parameters are passed as `message` argument with type `dict`. All parameters are stored in this `dict` by `name-value` pair. All params are named by it's Alterator originals (Basically as in `alterator-sh-functions`, but without `in_` prefix. So `$in__objects` here is `"_objects"`, `$in_action` is `"action"`, etc.)
|
||||
- `translate(text: str, domain=None)` - Output translated string. Optional param `domain` allows you to replace a default dictionary (`TEXTDOMAIN`) with your one. This is a wrapper over `gettext` utility. You may import this function as `_` to match `alterator-sh-functions` naming.
|
||||
- `_(text: str, domain=None)` - Output translated string. Optional param `domain` allows you to replace a default dictionary (`TEXTDOMAIN`) with your one. This is a wrapper over `gettext` utility.
|
||||
|
||||
#### Functions for processing of incoming parameters:
|
||||
- `test_bool(value)` - Incoming boolean variable are represented as a string with value `'#t'` or `'#f'`. This representation can be changed in the future and we are strongly recommend you to use `test_bool` function to test variable for `True` value.
|
||||
|
@ -1,44 +0,0 @@
|
||||
Name: alterator-python-functions
|
||||
|
||||
Version: 1.0.0
|
||||
Release: alt1
|
||||
|
||||
BuildRequires(pre): rpm-build-python3
|
||||
|
||||
BuildRequires: python3-module-setuptools
|
||||
|
||||
Requires: gettext
|
||||
|
||||
Conflicts: alterator < 3.4-alt1
|
||||
|
||||
BuildArch: noarch
|
||||
|
||||
Source: %name-%version.tar
|
||||
|
||||
Summary: Binding functions for Alterator Python based backends
|
||||
License: GPLv3
|
||||
Group: Development/Python3
|
||||
|
||||
%description
|
||||
Binding functions for Alterator Python based backends.
|
||||
Note that the module is `alterator_bindings.backend3`
|
||||
|
||||
%prep
|
||||
%setup -q
|
||||
|
||||
%build
|
||||
%pyproject_build
|
||||
|
||||
%check
|
||||
%pyproject_run_unittest
|
||||
|
||||
%install
|
||||
%pyproject_install
|
||||
|
||||
%files
|
||||
%python3_sitelibdir_noarch/alterator_bindings/*
|
||||
%doc README.md
|
||||
|
||||
%changelog
|
||||
* Fri Dec 27 2024 Sergey Konev <darisishe@altlinux.org> 1.0.0-alt1
|
||||
- Initial version
|
@ -31,12 +31,12 @@ import os
|
||||
from io import StringIO
|
||||
|
||||
ALTERATOR_DEBUG = False # May be set from backend for debug-messages
|
||||
TEXTDOMAIN = "" # Should be set by backend
|
||||
TEXTDOMAIN = None # Should be set by backend
|
||||
|
||||
|
||||
_LANGUAGE = "en_US" # Will be set from language parameter
|
||||
_OUT_BUF = StringIO() # Module-local variable
|
||||
_IN_MESSAGE_LOOP = False # Detects if we are already in message_loop()
|
||||
_OUT_BUF = None # Module-local variable
|
||||
|
||||
|
||||
### Internal function
|
||||
def _validate_symbol(str: str):
|
||||
@ -45,12 +45,8 @@ def _validate_symbol(str: str):
|
||||
|
||||
|
||||
### Quote
|
||||
def string_quote(str: str) -> str:
|
||||
"""Escapes \" and \\
|
||||
|
||||
Returns:
|
||||
str: The resulted string with escaped symbols.
|
||||
"""
|
||||
def string_quote(str: str):
|
||||
"""Escapes \" and \\"""
|
||||
escaped = str.translate(str.maketrans({"\\": r"\\", '"': r"\""}))
|
||||
return escaped
|
||||
|
||||
@ -63,11 +59,8 @@ def write_string(str: str):
|
||||
print(string_quote(str), end="", file=_OUT_BUF)
|
||||
|
||||
|
||||
def get_bool_string(value) -> str:
|
||||
"""Inteprets value as Scheme's boolean (value can be Python's bool or str)
|
||||
|
||||
Returns:
|
||||
str: '#t' or '#f' based on value"""
|
||||
def get_bool_string(value):
|
||||
"""Inteprets value as Scheme's boolean (value can be Python's bool or str)"""
|
||||
if (
|
||||
value == "yes"
|
||||
or value == "true"
|
||||
@ -81,11 +74,8 @@ def get_bool_string(value) -> str:
|
||||
return "#f"
|
||||
|
||||
|
||||
def test_bool(input: str) -> bool:
|
||||
"""Simple wrapper to cast Scheme's boolean to Python's
|
||||
|
||||
Returns:
|
||||
bool: True if input is '#t'"""
|
||||
def test_bool(input: str):
|
||||
"""Simple wrapper to cast Scheme's boolean to Python's"""
|
||||
return input == "#t"
|
||||
|
||||
|
||||
@ -104,11 +94,11 @@ def write_bool_param(name: str, value):
|
||||
print(name, get_bool_string(value), file=_OUT_BUF)
|
||||
|
||||
|
||||
def write_enum_item(name: str, label: str = ""):
|
||||
def write_enum_item(name: str, label: str = None):
|
||||
"""Writes enum entry (sets label value to name if not given)"""
|
||||
_validate_symbol(name)
|
||||
|
||||
if not label:
|
||||
if label is None:
|
||||
label = name
|
||||
|
||||
print('(name "{}" label "{}")'.format(name, string_quote(label)), file=_OUT_BUF)
|
||||
@ -152,12 +142,9 @@ def write_debug(message: str):
|
||||
### Localization Support
|
||||
|
||||
|
||||
def translate(text: str, domain="") -> str:
|
||||
"""Provides text translation from given domain (TEXTDOMAIN by default)
|
||||
|
||||
Returns:
|
||||
str: string-request to gettext to get required translation"""
|
||||
if not domain:
|
||||
def _(text: str, domain=None):
|
||||
"""Provides text translation from given domain (TEXTDOMAIN by default)"""
|
||||
if domain is None:
|
||||
domain = TEXTDOMAIN
|
||||
|
||||
lang_list = _LANGUAGE.split(":")
|
||||
@ -173,23 +160,24 @@ def translate(text: str, domain="") -> str:
|
||||
|
||||
|
||||
# Unimplemented for now
|
||||
def _check_response(response: str) -> bool:
|
||||
def _check_response(response: str):
|
||||
return True
|
||||
|
||||
|
||||
def _redirect_stdout() -> int:
|
||||
def _redirect_stdout():
|
||||
saved_stdout_fd = os.dup(sys.stdout.fileno())
|
||||
backendout = os.fdopen(saved_stdout_fd, "w")
|
||||
os.dup2(sys.stderr.fileno(), sys.stdout.fileno())
|
||||
return saved_stdout_fd
|
||||
return backendout
|
||||
|
||||
|
||||
def _ensure_textdomain():
|
||||
global TEXTDOMAIN
|
||||
if not TEXTDOMAIN:
|
||||
if TEXTDOMAIN is None:
|
||||
backend_name = os.path.splitext(os.path.basename(sys.argv[0]))[0]
|
||||
TEXTDOMAIN = "alterator-" + backend_name
|
||||
print(
|
||||
"alterator_bindings::backend3.py: TEXTDOMAIN variable is undefined! Setting to {}".format(
|
||||
"alterator_python_functions: TEXTDOMAIN variable is undefined! Setting to {}".format(
|
||||
TEXTDOMAIN
|
||||
),
|
||||
file=sys.stderr,
|
||||
@ -204,50 +192,39 @@ def _have_language(message: dict):
|
||||
|
||||
def message_loop(handler):
|
||||
"""Main function to communicate with Alterator"""
|
||||
|
||||
global _IN_MESSAGE_LOOP
|
||||
if _IN_MESSAGE_LOOP:
|
||||
sys.exit(
|
||||
"message_loop() called recursively! Redisign your on_message() callback, so it won't call message_loop() itself!"
|
||||
)
|
||||
|
||||
_IN_MESSAGE_LOOP = True
|
||||
|
||||
_ensure_textdomain()
|
||||
|
||||
# Redirecting streams (according to Alterator API)
|
||||
with os.fdopen(_redirect_stdout(), "w") as backendout:
|
||||
backendout = _redirect_stdout()
|
||||
|
||||
message = {}
|
||||
reading = False
|
||||
for line in sys.stdin:
|
||||
write_debug(">>>{}".format(line))
|
||||
message = {}
|
||||
reading = False
|
||||
for line in sys.stdin:
|
||||
write_debug(">>>{}".format(line))
|
||||
|
||||
if line == "_message:begin\n":
|
||||
message = {}
|
||||
reading = True
|
||||
if line == "_message:begin\n":
|
||||
message = {}
|
||||
reading = True
|
||||
|
||||
elif reading and line == "_message:end\n":
|
||||
_have_language(message)
|
||||
elif reading and line == "_message:end\n":
|
||||
_have_language(message)
|
||||
|
||||
reading = False
|
||||
global _OUT_BUF
|
||||
_OUT_BUF = StringIO()
|
||||
reading = False
|
||||
global _OUT_BUF
|
||||
_OUT_BUF = StringIO()
|
||||
|
||||
handler(message)
|
||||
response = _OUT_BUF.getvalue()
|
||||
handler(message)
|
||||
response = _OUT_BUF.getvalue()
|
||||
|
||||
# TODO: use _check_response here
|
||||
write_debug("response >>>({})<<<".format(response))
|
||||
print("({})".format(response), end="", file=backendout, flush=True)
|
||||
# TODO: use _check_response here
|
||||
write_debug("response >>>({})<<<".format(response))
|
||||
print("({})".format(response), end="", file=backendout, flush=True)
|
||||
|
||||
elif reading:
|
||||
name, value = line.split(":", 1)
|
||||
value = value.rstrip("\n")
|
||||
elif reading:
|
||||
name, value = line.split(":", 1)
|
||||
value = value.rstrip("\n")
|
||||
|
||||
value = re.sub(r"([^\\])\\n", r"\1\n", value)
|
||||
value = re.sub(r"\\\\", r"\\", value)
|
||||
value = re.sub(r"([^\\])\\n", r"\1\n", value)
|
||||
value = re.sub(r"\\\\", r"\\", value)
|
||||
|
||||
message[name] = value
|
||||
|
||||
_IN_MESSAGE_LOOP = False
|
||||
message[name] = value
|
@ -1,13 +0,0 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=70.1.0"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "alterator-python-functions"
|
||||
version = "1.0.0"
|
||||
description = "Binding functions for Alterator Python based backends"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.8"
|
||||
authors = [
|
||||
{ name="Sergey Konev", email="darisishe@altlinux.org" },
|
||||
]
|
@ -1,3 +0,0 @@
|
||||
"""Bindings to write alterator backends in Python."""
|
||||
|
||||
__version__ = "1.0.0"
|
@ -1,15 +0,0 @@
|
||||
import io
|
||||
from unittest import mock
|
||||
import tempfile
|
||||
|
||||
def mock_streams(func, input: str = ""):
|
||||
tmp = tempfile.NamedTemporaryFile()
|
||||
|
||||
with mock.patch("sys.stdin", new=io.StringIO(input)):
|
||||
with open(tmp.name, "w") as stdout:
|
||||
with mock.patch("sys.stdout", new=stdout):
|
||||
func()
|
||||
|
||||
with open(tmp.name, "r") as result:
|
||||
return result.read()
|
||||
|
@ -1,31 +0,0 @@
|
||||
from alterator_bindings.backend3 import *
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestBoolHelpersCase(unittest.TestCase):
|
||||
def test_get_true(self):
|
||||
self.assertEqual("#t", get_bool_string(True))
|
||||
self.assertEqual("#t", get_bool_string("yes"))
|
||||
self.assertEqual("#t", get_bool_string("#t"))
|
||||
self.assertEqual("#t", get_bool_string("y"))
|
||||
self.assertEqual("#t", get_bool_string("true"))
|
||||
self.assertEqual("#t", get_bool_string("1"))
|
||||
self.assertEqual("#t", get_bool_string("on"))
|
||||
|
||||
def test_get_false(self):
|
||||
self.assertEqual("#f", get_bool_string(False))
|
||||
self.assertEqual("#f", get_bool_string("no"))
|
||||
self.assertEqual("#f", get_bool_string("#f"))
|
||||
self.assertEqual("#f", get_bool_string("n"))
|
||||
self.assertEqual("#f", get_bool_string("TrUe"))
|
||||
self.assertEqual("#f", get_bool_string("fake"))
|
||||
self.assertEqual("#f", get_bool_string("off"))
|
||||
|
||||
def test_caster(self):
|
||||
self.assertTrue(test_bool("#t"))
|
||||
self.assertFalse(test_bool("#f"))
|
||||
self.assertFalse(test_bool("True"))
|
||||
self.assertFalse(test_bool("False"))
|
||||
self.assertFalse(test_bool("WTF"))
|
||||
|
@ -1,148 +0,0 @@
|
||||
from alterator_bindings import backend3
|
||||
from alterator_bindings.backend3 import *
|
||||
|
||||
from .mock_streams import mock_streams
|
||||
import unittest
|
||||
|
||||
|
||||
def on_message(_: dict):
|
||||
write_string("on_message")
|
||||
|
||||
|
||||
def on_message1(_):
|
||||
write_string("on_message1")
|
||||
|
||||
|
||||
def on_message2(message: dict):
|
||||
write_string("={}=".format(message.get("a", "")))
|
||||
|
||||
|
||||
def on_message3(message: dict):
|
||||
write_string("={}=".format(message.get("_objects", "")))
|
||||
|
||||
|
||||
class TestOnMessageCase(unittest.TestCase):
|
||||
def test_on_message1(self):
|
||||
input1 = """_message:begin
|
||||
_message:end
|
||||
"""
|
||||
|
||||
input2 = input1 + input1
|
||||
|
||||
expected2 = "(on_message1)(on_message1)"
|
||||
la = lambda: message_loop(on_message)
|
||||
self.assertEqual(
|
||||
"(on_message)", mock_streams(lambda: message_loop(on_message), input1)
|
||||
)
|
||||
self.assertEqual(
|
||||
"(on_message1)", mock_streams(lambda: message_loop(on_message1), input1)
|
||||
)
|
||||
self.assertEqual(
|
||||
expected2, mock_streams(lambda: message_loop(on_message1), input2)
|
||||
)
|
||||
|
||||
def test_on_message2(self):
|
||||
input1 = """_message:begin
|
||||
_message:begin
|
||||
_message:end
|
||||
"""
|
||||
|
||||
input2 = """_message:begin
|
||||
_message:end
|
||||
_message:end"""
|
||||
|
||||
input3 = """_message:begin
|
||||
a:b
|
||||
_message:end
|
||||
"""
|
||||
|
||||
input4 = """_message:begin
|
||||
a:b
|
||||
_message:begin
|
||||
_message:end
|
||||
"""
|
||||
|
||||
input5 = """_message:begin
|
||||
_message:end
|
||||
a:b
|
||||
_message:end
|
||||
"""
|
||||
|
||||
input6 = """_message:begin
|
||||
a:b
|
||||
_message:end
|
||||
_message:begin
|
||||
_message:end
|
||||
"""
|
||||
|
||||
expected6 = """(=b=)(==)"""
|
||||
|
||||
input7 = """_message:begin
|
||||
a:b
|
||||
a:c
|
||||
_message:end
|
||||
"""
|
||||
|
||||
self.assertEqual(
|
||||
"(==)",
|
||||
mock_streams(lambda: message_loop(on_message2), input1),
|
||||
"double message:begin",
|
||||
)
|
||||
self.assertEqual(
|
||||
"(==)",
|
||||
mock_streams(lambda: message_loop(on_message2), input2),
|
||||
"double message:end",
|
||||
)
|
||||
self.assertEqual(
|
||||
"(=b=)",
|
||||
mock_streams(lambda: message_loop(on_message2), input3),
|
||||
"parameter inside block",
|
||||
)
|
||||
self.assertNotEqual(
|
||||
"(=b=)",
|
||||
mock_streams(lambda: message_loop(on_message2), input4),
|
||||
"parameter before block",
|
||||
)
|
||||
self.assertNotEqual(
|
||||
"(=b=)",
|
||||
mock_streams(lambda: message_loop(on_message2), input5),
|
||||
"parameter after block",
|
||||
)
|
||||
self.assertEqual(
|
||||
expected6,
|
||||
mock_streams(lambda: message_loop(on_message2), input6),
|
||||
"parameter resetting",
|
||||
)
|
||||
self.assertEqual(
|
||||
"(=c=)",
|
||||
mock_streams(lambda: message_loop(on_message2), input7),
|
||||
"double parameter",
|
||||
)
|
||||
|
||||
def test_on_message3(self):
|
||||
input1 = """_message:begin
|
||||
_objects:test
|
||||
_message:end
|
||||
"""
|
||||
|
||||
input2 = """_message:begin
|
||||
_message:end
|
||||
"""
|
||||
|
||||
input3 = """
|
||||
_message:begin
|
||||
_objects:test\ntest
|
||||
_message:end
|
||||
"""
|
||||
|
||||
self.assertEqual(
|
||||
"(=test=)",
|
||||
mock_streams(lambda: message_loop(on_message3), input1),
|
||||
"_objects set",
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
"(==)",
|
||||
mock_streams(lambda: message_loop(on_message3), input2),
|
||||
"_objects unset",
|
||||
)
|
@ -1,29 +0,0 @@
|
||||
from alterator_bindings.backend3 import *
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestStringQuoteCase(unittest.TestCase):
|
||||
def test_simple(self):
|
||||
self.assertEqual(r"abcd", string_quote(r"abcd"))
|
||||
self.assertEqual(r"abc\\d", string_quote(r"abc\d"))
|
||||
self.assertEqual(r"ab\\c\\d", string_quote(r"ab\c\d"))
|
||||
self.assertEqual(r"abc\"d", string_quote(r'abc"d'))
|
||||
self.assertEqual(r"ab\"c\"d", string_quote(r'ab"c"d'))
|
||||
self.assertEqual(r"abc\\\"d", string_quote(r"abc\"d"))
|
||||
self.assertEqual(r"abcd", string_quote(r"abcd"))
|
||||
|
||||
def test_multiline(self):
|
||||
expected = r"""abcd
|
||||
abc\\d
|
||||
ab\\c\\d
|
||||
abc\"d
|
||||
ab\"c\"d"""
|
||||
|
||||
sample = r"""abcd
|
||||
abc\d
|
||||
ab\c\d
|
||||
abc"d
|
||||
ab"c"d"""
|
||||
|
||||
self.assertEqual(expected, string_quote(sample))
|
Reference in New Issue
Block a user