Commit 4e1f2881 authored by Thomas Gambier's avatar Thomas Gambier 🚴🏼

Update Release Candidate

parents 9085318c eec5a29e
...@@ -18,7 +18,7 @@ md5sum = d1e4d7306c39f2ebc64d0407860d4301 ...@@ -18,7 +18,7 @@ md5sum = d1e4d7306c39f2ebc64d0407860d4301
[template-cloudooo-instance] [template-cloudooo-instance]
filename = instance-cloudooo.cfg.in filename = instance-cloudooo.cfg.in
md5sum = 06dc19acd28ab412beffa61890be2095 md5sum = 1bc5c724d29337e1f35cfa60270fb08c
[template-haproxy-cfg] [template-haproxy-cfg]
filename = haproxy.cfg.in filename = haproxy.cfg.in
......
...@@ -20,6 +20,11 @@ ...@@ -20,6 +20,11 @@
"description": "The list of entry to add to the cloudooo mimetype registry. Each entry should on one line which format is: \"<source_mimetype> <destination_mimetype> <handler>\"", "description": "The list of entry to add to the cloudooo mimetype registry. Each entry should on one line which format is: \"<source_mimetype> <destination_mimetype> <handler>\"",
"textarea": true, "textarea": true,
"type": "string" "type": "string"
},
"enable-scripting": {
"description": "Enable the execution of scripts before saving a converted document. WARNING: Setting this parameter to true is unsafe, unless the CloudOoo server is private and not exposed to potential attackers",
"default": false,
"type": "boolean"
} }
} }
} }
...@@ -39,6 +39,11 @@ ...@@ -39,6 +39,11 @@
{% set apache_dict = {} -%} {% set apache_dict = {} -%}
{% do apache_dict.__setitem__(publish_url_name, (apache_port, "https", 'http://' ~ ipv4 ~ ':' ~ haproxy_port, False)) -%} {% do apache_dict.__setitem__(publish_url_name, (apache_port, "https", 'http://' ~ ipv4 ~ ':' ~ haproxy_port, False)) -%}
{% set ooo_enable_scripting = instance_parameter_dict['enable-scripting'] | string | lower -%}
{% if instance_parameter_dict.get('enable-scripting-parameter-name') -%}
{% set ooo_enable_scripting = slapparameter_dict.get(instance_parameter_dict['enable-scripting-parameter-name'], ooo_enable_scripting) | string | lower -%}
{% endif -%}
{% set bin_directory = parameter_dict['buildout-bin-directory'] -%} {% set bin_directory = parameter_dict['buildout-bin-directory'] -%}
{% set section_list = [] -%} {% set section_list = [] -%}
{% set cloudooo_section_list = [] -%} {% set cloudooo_section_list = [] -%}
...@@ -168,6 +173,7 @@ mimetype_entry_addition = ...@@ -168,6 +173,7 @@ mimetype_entry_addition =
ooo-binary-path = {{ parameter_dict['libreoffice-bin'] }}/program ooo-binary-path = {{ parameter_dict['libreoffice-bin'] }}/program
ooo-paster = {{ bin_directory }}/cloudooo_paster ooo-paster = {{ bin_directory }}/cloudooo_paster
ooo-uno-path = {{ parameter_dict['libreoffice-bin'] }}/basis-link/program ooo-uno-path = {{ parameter_dict['libreoffice-bin'] }}/basis-link/program
ooo_enable_scripting = {{ ooo_enable_scripting }}
{% for index in range(1, backend_count + 1) -%} {% for index in range(1, backend_count + 1) -%}
{% set name = 'cloudooo-' ~ index -%} {% set name = 'cloudooo-' ~ index -%}
......
...@@ -20,3 +20,5 @@ ssl-dict-parameter-name = ssl ...@@ -20,3 +20,5 @@ ssl-dict-parameter-name = ssl
mimetype-entry-addition-parameter-name = mimetype-entry-addition mimetype-entry-addition-parameter-name = mimetype-entry-addition
#mimetype-entry-addition = #mimetype-entry-addition =
# text/html application/pdf wkhtmltopdf # text/html application/pdf wkhtmltopdf
enable-scripting-parameter-name = enable-scripting
enable-scripting = false
...@@ -26,27 +26,28 @@ ...@@ -26,27 +26,28 @@
############################################################################## ##############################################################################
from setuptools import setup, find_packages from setuptools import setup, find_packages
version = '0.0.1.dev0' version = "0.0.1.dev0"
name = 'slapos.test.cloudooo' name = "slapos.test.cloudooo"
with open("README.md") as f: with open("README.md") as f:
long_description = f.read() long_description = f.read()
setup(name=name, setup(
name=name,
version=version, version=version,
description="Test for SlapOS' cloudooo", description="Test for SlapOS' CloudOoo",
long_description=long_description, long_description=long_description,
long_description_content_type='text/markdown', long_description_content_type="text/markdown",
maintainer="Nexedi", maintainer="Nexedi",
maintainer_email="info@nexedi.com", maintainer_email="info@nexedi.com",
url="https://lab.nexedi.com/nexedi/slapos", url="https://lab.nexedi.com/nexedi/slapos",
packages=find_packages(), packages=find_packages(),
install_requires=[ install_requires=[
'slapos.core', "slapos.core",
'slapos.cookbook', "slapos.cookbook",
'slapos.libnetworkcache', "slapos.libnetworkcache",
'requests', "requests",
'PyPDF2', "pypdf",
], ],
zip_safe=True, zip_safe=True,
test_suite='test', test_suite="test",
) )
...@@ -24,324 +24,550 @@ ...@@ -24,324 +24,550 @@
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
# #
############################################################################## ##############################################################################
import base64
import codecs import codecs
import csv import csv
import multiprocessing import io
import os
import json import json
import xmlrpc.client as xmlrpclib import multiprocessing
import urllib.parse as urllib_parse
import ssl import ssl
import base64 import textwrap
import io import urllib.parse as urllib_parse
import xmlrpc.client as xmlrpclib
from functools import partial
from pathlib import Path
from typing import AbstractSet, Callable, ClassVar, Dict, Iterable, Mapping
import requests
import PIL.Image import PIL.Image
import PyPDF2 import pypdf
import requests
from slapos.testing.testcase import makeModuleSetUpAndTestCaseClass from slapos.testing.testcase import makeModuleSetUpAndTestCaseClass
from slapos.testing.utils import ImageComparisonTestCase from slapos.testing.utils import ImageComparisonTestCase
setUpModule, _CloudOooTestCase = makeModuleSetUpAndTestCaseClass( setUpModule, _CloudOooTestCase = makeModuleSetUpAndTestCaseClass(
os.path.abspath( str((Path(__file__).parent.parent / "software.cfg").absolute())
os.path.join(os.path.dirname(__file__), '..', 'software.cfg'))) )
class CloudOooTestCase(_CloudOooTestCase): def open_cloudooo_connection(url):
# Cloudooo needs a lot of time before being available. """
instance_max_retry = 30 Open a RPC connection with Cloudooo.
def setUp(self): Args:
self.url = json.loads( url: The URL of the CloudOoo server.
self.computer_partition.getConnectionParameterDict()["_"])['cloudooo']
Returns:
A object that manages communication with CloudOoo via XML-RCP.
"""
# XXX ignore certificate errors # XXX ignore certificate errors
ssl_context = ssl.create_default_context() ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE ssl_context.verify_mode = ssl.CERT_NONE
self.server = xmlrpclib.ServerProxy( return xmlrpclib.ServerProxy(
self.url, url,
context=ssl_context, context=ssl_context,
allow_none=True, allow_none=True,
) )
self.addCleanup(self.server('close'))
def normalizeFontName(font_name): def convert_file(
if '+' in font_name: file,
return font_name.split('+')[1] source_format,
if font_name.startswith('/'): destination_format,
return font_name[1:] *,
zip=False,
refresh=False,
conversion_kw={},
server,
):
"""
Converts a file using CloudOoo.
This is a helper function that does the necessary encoding/decoding,
providing type-safety and keyword arguments.
Args:
file: The file contents to send to CloudOoo.
source_format: Format of the input file.
destination_format: Format of the output file.
zip: Whether a zip file should be returned.
refresh: Whether dynamic properties of document should be replaced
before conversion.
conversion_kw: Additional arguments for the conversion.
server: The server used to send requests to Cloudooo.
Returns:
Contents of the converted file.
def getReferencedFonts(pdf_file_reader):
"""Return fonts referenced in this pdf
""" """
fonts = set() converted_file = server.convertFile(
base64.encodebytes(file).decode(),
source_format,
destination_format,
zip,
refresh,
conversion_kw,
)
assert isinstance(converted_file, str)
return base64.decodebytes(converted_file.encode())
def collectFonts(obj):
"""Recursively visit PDF objects and collect referenced fonts in `fonts` class CloudOooTestCase(_CloudOooTestCase):
""" """
if hasattr(obj, 'keys'): Parent class for all CloudOoo tests.
if '/BaseFont' in obj:
fonts.add(obj['/BaseFont']) This class sets some attributes in the `setUp` method that are necessary
for k in obj.keys(): for testing CloudOoo.
collectFonts(obj[k])
Attributes:
url: The URL of the CloudOoo server.
server: A object that manages communication with CloudOoo via XML-RCP.
"""
# Cloudooo needs a lot of time before being available.
instance_max_retry = 30
def setUp(self):
self.url = json.loads(self.computer_partition.getConnectionParameterDict()["_"])[
"cloudooo"
]
self.server = open_cloudooo_connection(self.url)
self.addCleanup(self.server("close"))
def convert_file(
self,
file,
source_format,
destination_format,
*,
zip=False,
refresh=False,
conversion_kw={},
):
"""
Converts a file using CloudOoo.
This is a helper method that does the necessary encoding/decoding,
providing type-safety and keyword arguments.
Args:
file: The file contents to send to CloudOoo.
source_format: Format of the input file.
destination_format: Format of the output file.
zip: Whether a zip file should be returned.
refresh: Whether dynamic properties of document should be replaced
before conversion.
conversion_kw: Additional arguments for the conversion.
Returns:
Contents of the converted file.
"""
return convert_file(
file=file,
source_format=source_format,
destination_format=destination_format,
zip=zip,
refresh=refresh,
conversion_kw=conversion_kw,
server=self.server,
)
def script_test_basic(self):
"""
Tries to execute a hello world script.
Returns:
The file contents in base64. If the script is executed
properly, it should contain the string ``"Hello World"``,
preceded by the UTF-8 BOM and with a trailing newline.
"""
script = textwrap.dedent(
"""\
# Get the XText interface
text = Document.Text
# Create an XTextRange at the end of the document
tRange = text.End
# Set the string
tRange.String = "Hello World"
""",
)
return self.convert_file(
b"<html></html>",
"html",
"txt",
conversion_kw={"script": script},
)
QT_FONT_MAPPING = {
"Arial": "LiberationSans",
"Arial Black": "LiberationSans",
"Avant Garde": "LiberationSans",
"Bookman": "LiberationSans",
"Carlito": "Carlito",
"Comic Sans MS": "LiberationSans",
"Courier New": "LiberationSans",
"DejaVu Sans": "DejaVuSans",
"DejaVu Sans Condensed": "LiberationSans",
"DejaVu Sans Mono": "DejaVuSansMono",
"DejaVu Serif": "DejaVuSerif",
"DejaVu Serif Condensed": "LiberationSans",
"Garamond": "LiberationSans",
"Gentium Basic": "GentiumBasic",
"Gentium Book Basic": "GentiumBookBasic",
"Georgia": "LiberationSans",
"Helvetica": "LiberationSans",
"IPAex Gothic": "LiberationSans",
"IPAex Mincho": "LiberationSans",
"Impact": "LiberationSans",
"Liberation Mono": "LiberationMono",
"Liberation Sans": "LiberationSans",
"Liberation Sans Narrow": "LiberationSansNarrow",
"Liberation Serif": "LiberationSerif",
"Linux LibertineG": "LiberationSans",
"OpenSymbol": {"NotoSans-Regular", "OpenSymbol"},
"Palatino": "LiberationSans",
"Roboto Black": "LiberationSans",
"Roboto Condensed Light": "LiberationSans",
"Roboto Condensed": "RobotoCondensed-Regular",
"Roboto Light": "LiberationSans",
"Roboto Medium": "LiberationSans",
"Roboto Thin": "LiberationSans",
"Times New Roman": "LiberationSans",
"Trebuchet MS": "LiberationSans",
"Verdana": "LiberationSans",
"ZZZdefault fonts when no match": "LiberationSans",
}
LIBREOFFICE_FONT_MAPPING = {
"Arial": "LiberationSans",
"Arial Black": "NotoSans-Regular",
"Avant Garde": "NotoSans-Regular",
"Bookman": "NotoSans-Regular",
"Carlito": "Carlito",
"Comic Sans MS": "NotoSans-Regular",
"Courier New": "LiberationMono",
"DejaVu Sans": "DejaVuSans",
"DejaVu Sans Condensed": "DejaVuSansCondensed",
"DejaVu Sans Mono": "DejaVuSansMono",
"DejaVu Serif": "DejaVuSerif",
"DejaVu Serif Condensed": "DejaVuSerifCondensed",
"Garamond": "NotoSerif-Regular",
"Gentium Basic": "GentiumBasic",
"Gentium Book Basic": "GentiumBookBasic",
"Georgia": "NotoSerif-Regular",
"Helvetica": "LiberationSans",
"IPAex Gothic": "IPAexGothic",
"IPAex Mincho": "IPAexMincho",
"Impact": "NotoSans-Regular",
"Liberation Mono": "LiberationMono",
"Liberation Sans": "LiberationSans",
"Liberation Sans Narrow": "LiberationSansNarrow",
"Liberation Serif": "LiberationSerif",
"Linux LibertineG": "LinuxLibertineG",
"OpenSymbol": {"OpenSymbol", "IPAMincho"},
"Palatino": "NotoSerif-Regular",
"Roboto Black": "Roboto-Black",
"Roboto Condensed Light": "RobotoCondensed-Light",
"Roboto Condensed": "RobotoCondensed-Regular",
"Roboto Light": "Roboto-Light",
"Roboto Medium": "Roboto-Medium",
"Roboto Thin": "Roboto-Thin",
"Times New Roman": "LiberationSerif",
"Trebuchet MS": "NotoSans-Regular",
"Verdana": "NotoSans-Regular",
"ZZZdefault fonts when no match": "NotoSans-Regular",
}
def normalize_font_name(font_name):
"""
Normalize a font name.
As with other PostScript markup, font names are written with a leading
slash symbol ("/"), which has to be stripped to obtain the conventional font
name.
Moreover, the standard allows also to define "font subsets", for which a tag
followed by a plus sign ("+") precedes the actual font name:
https://opensource.adobe.com/dc-acrobat-sdk-docs/pdfstandards/PDF32000_2008.pdf#page=266
This tag is also removed by this function.
Args:
font_name: The font name to normalize.
Returns:
Normalized font name.
"""
if "+" in font_name:
return font_name.split("+")[1]
for page in pdf_file_reader.pages: if font_name.startswith("/"):
collectFonts(page.getObject()['/Resources']) return font_name[1:]
raise ValueError("Invalid font name")
return {normalizeFontName(font) for font in fonts}
def get_referenced_fonts(
pdf_file_reader,
):
"""
Return fonts referenced in a pdf.
class HTMLtoPDFConversionFontTestMixin: Returns a set with all font names (normalized) present in a PDF.
"""Mix-In class to test how fonts are selected during
HTML to PDF conversions.
This needs to be mixed with a test case defining: Args:
pdf_file_reader: PDF reader.
Returns:
Set of font names present in the PDF.
* pdf_producer : the name of /Producer in PDF metadata
* expected_font_mapping : a mapping of resulting font name in pdf,
keyed by font-family in the input html
* _convert_html_to_pdf: a method to to convert html to pdf
""" """
def _convert_html_to_pdf(self, src_html):
# type: (str) -> bytes def fonts_in_obj(obj):
"""Convert the HTML source to pdf bytes.
""" """
Yield fonts from a PDF object.
def test(self): Recursively visit PDF objects and yield referenced fonts in `fonts`.
Args:
obj: A reference to a PDF object.
Yields:
The font names in the object.
"""
if hasattr(obj, "keys"):
if "/BaseFont" in obj:
yield obj["/BaseFont"]
for k in obj.keys():
yield from fonts_in_obj(obj[k])
return {
normalize_font_name(font)
for page in pdf_file_reader.pages
for font in fonts_in_obj(page.get_object()["/Resources"])
}
class TestDefaultInstance(CloudOooTestCase, ImageComparisonTestCase):
"""Tests for CloudOoo instance with default configuration."""
__partition_reference__ = "co_default"
def assert_pdf_conversion_metadata(
self,
convert_html_to_pdf,
*,
expected_producer,
expected_font_mapping,
):
actual_font_mapping_mapping = {} actual_font_mapping_mapping = {}
for font in self.expected_font_mapping:
src_html = f''' for font in expected_font_mapping:
src_html = f"""
<style> <style>
p {{ font-family: "{font}"; font-size: 20pt; }} p {{ font-family: "{font}"; font-size: 20pt; }}
</style> </style>
<p>the quick brown fox jumps over the lazy dog.</p> <p>the quick brown fox jumps over the lazy dog.</p>
<p>THE QUICK BROWN FOX JUMPS OVER THE LAZY DOG.</p> <p>THE QUICK BROWN FOX JUMPS OVER THE LAZY DOG.</p>
''' """
pdf_data = convert_html_to_pdf(src_html.encode())
pdf_reader = pypdf.PdfReader(io.BytesIO(pdf_data))
metadata = pdf_reader.metadata
assert metadata
pdf_data = self._convert_html_to_pdf(src_html)
pdf_reader = PyPDF2.PdfFileReader(io.BytesIO(pdf_data))
self.assertEqual( self.assertEqual(
self.pdf_producer, metadata.producer,
pdf_reader.getDocumentInfo()['/Producer']) expected_producer,
fonts_in_pdf = getReferencedFonts(pdf_reader) )
fonts_in_pdf = get_referenced_fonts(pdf_reader)
font_or_set = fonts_in_pdf
if len(fonts_in_pdf) == 1: if len(fonts_in_pdf) == 1:
actual_font_mapping_mapping[font] = fonts_in_pdf.pop() # Tuple unpacking
else: (font_or_set,) = fonts_in_pdf
actual_font_mapping_mapping[font] = fonts_in_pdf
self.maxDiff = None
self.assertEqual(self.expected_font_mapping, actual_font_mapping_mapping)
class TestWkhtmlToPDF(HTMLtoPDFConversionFontTestMixin, CloudOooTestCase):
__partition_reference__ = 'wk'
pdf_producer = 'Qt 4.8.7'
expected_font_mapping = {
'Arial': 'LiberationSans',
'Arial Black': 'LiberationSans',
'Avant Garde': 'LiberationSans',
'Bookman': 'LiberationSans',
'Carlito': 'Carlito',
'Comic Sans MS': 'LiberationSans',
'Courier New': 'LiberationSans',
'DejaVu Sans': 'DejaVuSans',
'DejaVu Sans Condensed': 'LiberationSans',
'DejaVu Sans Mono': 'DejaVuSansMono',
'DejaVu Serif': 'DejaVuSerif',
'DejaVu Serif Condensed': 'LiberationSans',
'Garamond': 'LiberationSans',
'Gentium Basic': 'GentiumBasic',
'Gentium Book Basic': 'GentiumBookBasic',
'Georgia': 'LiberationSans',
'Helvetica': 'LiberationSans',
'IPAex Gothic': 'LiberationSans',
'IPAex Mincho': 'LiberationSans',
'Impact': 'LiberationSans',
'Liberation Mono': 'LiberationMono',
'Liberation Sans': 'LiberationSans',
'Liberation Sans Narrow': 'LiberationSansNarrow',
'Liberation Serif': 'LiberationSerif',
'Linux LibertineG': 'LiberationSans',
'OpenSymbol': {'NotoSans-Regular', 'OpenSymbol'},
'Palatino': 'LiberationSans',
'Roboto Black': 'LiberationSans',
'Roboto Condensed Light': 'LiberationSans',
'Roboto Condensed': 'RobotoCondensed-Regular',
'Roboto Light': 'LiberationSans',
'Roboto Medium': 'LiberationSans',
'Roboto Thin': 'LiberationSans',
'Times New Roman': 'LiberationSans',
'Trebuchet MS': 'LiberationSans',
'Verdana': 'LiberationSans',
'ZZZdefault fonts when no match': 'LiberationSans'
}
def _convert_html_to_pdf(self, src_html): actual_font_mapping_mapping[font] = font_or_set
return base64.decodebytes(
self.server.convertFile(
base64.encodebytes(src_html.encode()).decode(),
'html',
'pdf',
False,
False,
{
'encoding': 'utf-8'
},
).encode())
class TestLibreoffice(HTMLtoPDFConversionFontTestMixin, CloudOooTestCase):
__partition_reference__ = 'lo'
pdf_producer = 'LibreOffice 7.5'
expected_font_mapping = {
'Arial': 'LiberationSans',
'Arial Black': 'NotoSans-Regular',
'Avant Garde': 'NotoSans-Regular',
'Bookman': 'NotoSans-Regular',
'Carlito': 'Carlito',
'Comic Sans MS': 'NotoSans-Regular',
'Courier New': 'LiberationMono',
'DejaVu Sans': 'DejaVuSans',
'DejaVu Sans Condensed': 'DejaVuSansCondensed',
'DejaVu Sans Mono': 'DejaVuSansMono',
'DejaVu Serif': 'DejaVuSerif',
'DejaVu Serif Condensed': 'DejaVuSerifCondensed',
'Garamond': 'NotoSerif-Regular',
'Gentium Basic': 'GentiumBasic',
'Gentium Book Basic': 'GentiumBookBasic',
'Georgia': 'NotoSerif-Regular',
'Helvetica': 'LiberationSans',
'IPAex Gothic': 'IPAexGothic',
'IPAex Mincho': 'IPAexMincho',
'Impact': 'NotoSans-Regular',
'Liberation Mono': 'LiberationMono',
'Liberation Sans': 'LiberationSans',
'Liberation Sans Narrow': 'LiberationSansNarrow',
'Liberation Serif': 'LiberationSerif',
'Linux LibertineG': 'LinuxLibertineG',
'OpenSymbol': {'OpenSymbol', 'IPAMincho'},
'Palatino': 'NotoSerif-Regular',
'Roboto Black': 'Roboto-Black',
'Roboto Condensed Light': 'RobotoCondensed-Light',
'Roboto Condensed': 'RobotoCondensed-Regular',
'Roboto Light': 'Roboto-Light',
'Roboto Medium': 'Roboto-Medium',
'Roboto Thin': 'Roboto-Thin',
'Times New Roman': 'LiberationSerif',
'Trebuchet MS': 'NotoSans-Regular',
'Verdana': 'NotoSans-Regular',
'ZZZdefault fonts when no match': 'NotoSans-Regular'
}
def _convert_html_to_pdf(self, src_html): self.assertEqual(actual_font_mapping_mapping, expected_font_mapping)
return base64.decodebytes(
self.server.convertFile( def html_to_pdf_wkhtmltopdf_convert(self, src_html):
base64.encodebytes(src_html.encode()).decode(), """HTML to PDF conversion using wkhtmltopdf."""
'html', return self.convert_file(
'pdf', src_html,
).encode()) "html",
"pdf",
conversion_kw={"encoding": "utf-8"},
class TestLibreofficeDrawToPNGConversion(CloudOooTestCase, ImageComparisonTestCase): )
__partition_reference__ = 'l'
def test_html_to_pdf_wkhtmltopdf(self):
def test(self): """Test HTML to PDF conversion using wkhtmltopdf."""
reference_png = PIL.Image.open(os.path.join('data', f'{self.id()}.png')) self.assert_pdf_conversion_metadata(
with open(os.path.join('data', f'{self.id()}.odg'), 'rb') as f: self.html_to_pdf_wkhtmltopdf_convert,
actual_png_data = base64.decodebytes( expected_producer="Qt 4.8.7",
self.server.convertFile( expected_font_mapping=QT_FONT_MAPPING,
base64.encodebytes(f.read()).decode(), )
'odg',
'png', def html_to_pdf_libreoffice_convert(self, src_html):
).encode()) """HTML to PDF conversion using LibreOffice."""
return self.convert_file(
src_html,
"html",
"pdf",
)
def test_html_to_pdf_libreoffice_convert(self):
"""Test HTML to PDF conversion using wkhtmltopdf."""
self.assert_pdf_conversion_metadata(
self.html_to_pdf_libreoffice_convert,
expected_producer="LibreOffice 7.5",
expected_font_mapping=LIBREOFFICE_FONT_MAPPING,
)
def test_draw_to_png(self):
"""Test Draw's ODG to PNG conversion."""
reference_png = PIL.Image.open("data/test_draw_to_png.png")
with open("data/test_draw_to_png.odg", "rb") as f:
actual_png_data = self.convert_file(
f.read(),
"odg",
"png",
)
actual_png = PIL.Image.open(io.BytesIO(actual_png_data)) actual_png = PIL.Image.open(io.BytesIO(actual_png_data))
# save a snapshot # Save a snapshot
with open(os.path.join(self.computer_partition_root_path, self.id() + '.png'), 'wb') as f: with open(
Path(self.computer_partition_root_path) / "test_draw_to_png.png",
"wb",
) as f:
f.write(actual_png_data) f.write(actual_png_data)
self.assertImagesSame(actual_png, reference_png) self.assertImagesSame(actual_png, reference_png)
class TestLibreOfficeTextConversion(CloudOooTestCase):
__partition_reference__ = 'txt'
def test_html_to_text(self): def test_html_to_text(self):
"""Test HTML to TXT conversion."""
file_content = self.convert_file(
"<html>héhé</html>".encode(),
"html",
"txt",
)
self.assertEqual( self.assertEqual(
base64.decodebytes( file_content,
self.server.convertFile( codecs.BOM_UTF8 + b"h\xc3\xa9h\xc3\xa9\n",
base64.encodebytes( )
'<html>héhé</html>'.encode()).decode(),
'html', def test_scripting_disabled(self):
'txt', """Test that the basic script raises when scripting is disabled."""
).encode()), with self.assertRaisesRegex(Exception, "ooo: scripting is disabled"):
codecs.BOM_UTF8 + b'h\xc3\xa9h\xc3\xa9\n', self.script_test_basic()
def _convert_html_to_text(
src_html,
*,
url,
):
"""
Convert HTML to TXT.
This is a helper method for using with map.
Args:
src_html: HTML to convert.
url: URL of the CloudOoo server.
Returns:
Converted file contents.
"""
with open_cloudooo_connection(url) as server:
return convert_file(
src_html,
"html",
"txt",
server=server,
) )
class TestLibreOfficeCluster(CloudOooTestCase): class TestLibreOfficeCluster(CloudOooTestCase):
__partition_reference__ = 'lc' """Class for testing a cluster with multiple backends."""
__partition_reference__ = "co_cluster"
@classmethod @classmethod
def getInstanceParameterDict(cls): def getInstanceParameterDict(cls):
return {'backend-count': 4} return {"backend-count": 4}
def test_multiple_conversions(self): def test_multiple_conversions(self):
# make this function global so that it can be picked and used by multiprocessing """Test that concurrent requests are distributed in the cluster."""
global _convert_html_to_text
def _convert_html_to_text(src_html):
return base64.decodebytes(
self.server.convertFile(
base64.encodebytes(src_html.encode()).decode(),
'html',
'txt',
).encode())
pool = multiprocessing.Pool(5) pool = multiprocessing.Pool(5)
with pool: with pool:
converted = pool.map( converted = pool.map(
_convert_html_to_text, partial(_convert_html_to_text, url=self.url),
['<html><body>hello</body></html>'] * 100) [b"<html><body>hello</body></html>"] * 100,
)
self.assertEqual(converted, [codecs.BOM_UTF8 + b'hello\n'] * 100) self.assertEqual(converted, [codecs.BOM_UTF8 + b"hello\n"] * 100)
# haproxy stats are exposed # Haproxy stats are exposed
res = requests.get( res = requests.get(
urllib_parse.urljoin(self.url, '/haproxy;csv'), urllib_parse.urljoin(self.url, "/haproxy;csv"),
verify=False, verify=False,
) )
reader = csv.DictReader(io.StringIO(res.text)) reader = csv.DictReader(io.StringIO(res.text))
line_list = list(reader) line_list = list(reader)
# requests have been balanced # Requests have been balanced
total_hrsp_2xx = { total_hrsp_2xx = {line["svname"]: int(line["hrsp_2xx"]) for line in line_list}
line['svname']: int(line['hrsp_2xx']) self.assertEqual(total_hrsp_2xx["FRONTEND"], 100)
for line in line_list self.assertEqual(total_hrsp_2xx["BACKEND"], 100)
} for backend in "cloudooo_1", "cloudooo_2", "cloudooo_3", "cloudooo_4":
self.assertEqual(total_hrsp_2xx['FRONTEND'], 100) # Ideally there should be 25% of requests on each backend, because we use
self.assertEqual(total_hrsp_2xx['BACKEND'], 100)
for backend in 'cloudooo_1', 'cloudooo_2', 'cloudooo_3', 'cloudooo_4':
# ideally there should be 25% of requests on each backend, because we use
# round robin scheduling, but it can happen that some backend take longer # round robin scheduling, but it can happen that some backend take longer
# to start, so we are tolerant here and just check that each backend # to start, so we are tolerant here and just check that each backend
# process at least one request. # process at least one request.
self.assertGreater(total_hrsp_2xx[backend], 0) self.assertGreater(total_hrsp_2xx[backend], 0)
# no errors # No errors
total_eresp = { total_eresp = {line["svname"]: int(line["eresp"] or 0) for line in line_list}
line['svname']: int(line['eresp'] or 0) self.assertEqual(
for line in line_list total_eresp,
} {
"FRONTEND": 0,
"cloudooo_1": 0,
"cloudooo_2": 0,
"cloudooo_3": 0,
"cloudooo_4": 0,
"BACKEND": 0,
},
)
class TestLibreOfficeScripting(CloudOooTestCase):
"""Class with scripting enabled, to try that functionality."""
__partition_reference__ = "co_script"
@classmethod
def getInstanceParameterDict(cls):
"""Enable scripting for this instance."""
return {"enable-scripting": True}
def test_scripting_basic(self):
"""Test that the basic script works."""
file = self.script_test_basic()
self.assertEqual( self.assertEqual(
total_eresp, { file,
'FRONTEND': 0, codecs.BOM_UTF8 + b"Hello World\n",
'cloudooo_1': 0, )
'cloudooo_2': 0,
'cloudooo_3': 0,
'cloudooo_4': 0,
'BACKEND': 0,
})
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
[template] [template]
filename = instance.cfg filename = instance.cfg
md5sum = 2e30c07c6436895ac0bc6c177cf7013d md5sum = f096f3cbb414730a9f0d46b0c0f5eb22
[template-ors] [template-ors]
filename = instance-ors.cfg filename = instance-ors.cfg
...@@ -24,7 +24,7 @@ md5sum = f5c76c3443b75569eb18503dce38e783 ...@@ -24,7 +24,7 @@ md5sum = f5c76c3443b75569eb18503dce38e783
[slaplte.jinja2] [slaplte.jinja2]
_update_hash_filename_ = slaplte.jinja2 _update_hash_filename_ = slaplte.jinja2
md5sum = 871ade334f445e22d6cb473e4d4e3522 md5sum = 27c49897c9a3e4c260105534bed0132d
[ru_amarisoft-stats.jinja2.py] [ru_amarisoft-stats.jinja2.py]
_update_hash_filename_ = ru/amarisoft-stats.jinja2.py _update_hash_filename_ = ru/amarisoft-stats.jinja2.py
...@@ -36,44 +36,16 @@ md5sum = ab666fdfadbfc7d8a16ace38d295c883 ...@@ -36,44 +36,16 @@ md5sum = ab666fdfadbfc7d8a16ace38d295c883
[ru_libinstance.jinja2.cfg] [ru_libinstance.jinja2.cfg]
_update_hash_filename_ = ru/libinstance.jinja2.cfg _update_hash_filename_ = ru/libinstance.jinja2.cfg
md5sum = 2dda7713832be83d94522c7abb4901f9 md5sum = 7613c4decd4468ab5d826421aef955f1
[ru_sdr_libinstance.jinja2.cfg] [ru_sdr_libinstance.jinja2.cfg]
_update_hash_filename_ = ru/sdr/libinstance.jinja2.cfg _update_hash_filename_ = ru/sdr/libinstance.jinja2.cfg
md5sum = b7906ca3a6b17963f78f680fc0842b74 md5sum = b7906ca3a6b17963f78f680fc0842b74
[ru_lopcomm_libinstance.jinja2.cfg]
_update_hash_filename_ = ru/lopcomm/libinstance.jinja2.cfg
md5sum = c3bd882559ab9cd2a068519ea5d8c92e
[ru_sunwave_libinstance.jinja2.cfg] [ru_sunwave_libinstance.jinja2.cfg]
_update_hash_filename_ = ru/sunwave/libinstance.jinja2.cfg _update_hash_filename_ = ru/sunwave/libinstance.jinja2.cfg
md5sum = bc5d82b8737b6990674b280ef2774be7 md5sum = bc5d82b8737b6990674b280ef2774be7
[ru_lopcomm_ncclient_common.py]
_update_hash_filename_ = ru/lopcomm/ncclient_common.py
md5sum = 8dbe6a48fc0fca4f0cbd0c746be1aeda
[ru_lopcomm_stats.jinja2.py]
_update_hash_filename_ = ru/lopcomm/stats.jinja2.py
md5sum = b7ec0025a92e0947e4ac6abc4b06bf19
[ru_lopcomm_config.jinja2.py]
_update_hash_filename_ = ru/lopcomm/config.jinja2.py
md5sum = 122726666d147447171dcae9ebf8d093
[ru_lopcomm_reset-info.jinja2.py]
_update_hash_filename_ = ru/lopcomm/reset-info.jinja2.py
md5sum = 3d78df1993211efaabd3dc6f2ec8de30
[ru_lopcomm_reset.jinja2.py]
_update_hash_filename_ = ru/lopcomm/reset.jinja2.py
md5sum = 9741fbc99aaf768e9cc3ab48925dfee5
[ru_lopcomm_software.jinja2.py]
_update_hash_filename_ = ru/lopcomm/software.jinja2.py
md5sum = 2b08bb666c5f3ab287cdddbfdb4c9249
[ru_tapsplit] [ru_tapsplit]
_update_hash_filename_ = ru/tapsplit _update_hash_filename_ = ru/tapsplit
md5sum = 700aab566289619fb83ac6f3b085d983 md5sum = 700aab566289619fb83ac6f3b085d983
...@@ -88,7 +60,7 @@ md5sum = 52da9fe3a569199e35ad89ae1a44c30e ...@@ -88,7 +60,7 @@ md5sum = 52da9fe3a569199e35ad89ae1a44c30e
[template-enb] [template-enb]
_update_hash_filename_ = instance-enb.jinja2.cfg _update_hash_filename_ = instance-enb.jinja2.cfg
md5sum = 8b9301f26fc4ffbc7eda9c1ac8da1a46 md5sum = 140cce72eb04768abbe87ea4982f36bd
[template-ors-enb] [template-ors-enb]
_update_hash_filename_ = instance-ors-enb.jinja2.cfg _update_hash_filename_ = instance-ors-enb.jinja2.cfg
...@@ -150,14 +122,6 @@ md5sum = f07c85916bcb7e4002c8edc3d087c1be ...@@ -150,14 +122,6 @@ md5sum = f07c85916bcb7e4002c8edc3d087c1be
filename = config/ue.jinja2.cfg filename = config/ue.jinja2.cfg
md5sum = 62291a11fd36a42464901cdc81338687 md5sum = 62291a11fd36a42464901cdc81338687
[ru_lopcomm_CreateProcessingEle.jinja2.xml]
_update_hash_filename_ = ru/lopcomm/CreateProcessingEle.jinja2.xml
md5sum = e435990eb0a0d4be41efa9bd16dce09b
[ru_lopcomm_cu_config.jinja2.xml]
_update_hash_filename_ = ru/lopcomm/cu_config.jinja2.xml
md5sum = 346c911e1ac5e5001a39c8926b44c91e
[software.cfg.html] [software.cfg.html]
_update_hash_filename_ = gadget/software.cfg.html _update_hash_filename_ = gadget/software.cfg.html
md5sum = 61a2f783fbf683a34aed3d13e00baca2 md5sum = 61a2f783fbf683a34aed3d13e00baca2
......
...@@ -105,9 +105,6 @@ ...@@ -105,9 +105,6 @@
{ {
"$ref": "../ru/sdr/input-schema.json" "$ref": "../ru/sdr/input-schema.json"
}, },
{
"$ref": "../ru/lopcomm/input-schema.json"
},
{ {
"$ref": "../ru/sunwave/input-schema.json" "$ref": "../ru/sunwave/input-schema.json"
} }
......
...@@ -198,18 +198,18 @@ ...@@ -198,18 +198,18 @@
"default": 0 "default": 0
}, },
"xlog_fluentbit_forward_host": { "xlog_fluentbit_forward_host": {
"title": "Address to Forward Xlog by Fluenbit", "title": "Fluentbit Xlog forwarding address",
"description": "Address of Remote Fluentd or Fluentbit Server to Forward Xlog", "description": "Address of remote Fluentd or Fluentbit server to which Fluentbit should forward Xlog data",
"type": "string" "type": "string"
}, },
"xlog_fluentbit_forward_port": { "xlog_fluentbit_forward_port": {
"title": "Port to Forward Xlog by Fluentbit", "title": "Fluentbit Xlog forwarding port",
"description": "Optional Port of Remote Fluentd or Fluentbit Server to Forward Xlog", "description": "(Optional) Port of remote Fluentd or Fluentbit server to which Fluentbit should forward Xlog data",
"type": "string" "type": "string"
}, },
"xlog_fluentbit_forward_shared_key": { "xlog_fluentbit_forward_shared_key": {
"title": "Shared Key to Forward Xlog by Fluentbit", "title": "Fluentbit Xlog forwarding shared key",
"description": "Secret Key Shared with Remote Fluentd or Fluentbit Server for Authentication when Forwarding Xlog", "description": "Secret Key shared with remote Fluentd or Fluentbit server for authentication when forwarding Xlog data",
"type": "string" "type": "string"
} }
} }
......
...@@ -80,6 +80,7 @@ script = ${:etc}/run ...@@ -80,6 +80,7 @@ script = ${:etc}/run
service = ${:etc}/service service = ${:etc}/service
promise = ${:etc}/promise promise = ${:etc}/promise
log = ${:var}/log log = ${:var}/log
xlog-fluentbit = ${:var}/xlog-fluentbit
{% if slapparameter_dict.get("enb_config_link", None) %} {% if slapparameter_dict.get("enb_config_link", None) %}
[enb-config-dl] [enb-config-dl]
...@@ -149,6 +150,26 @@ command-line = ${xamari-xlog-script:output} ...@@ -149,6 +150,26 @@ command-line = ${xamari-xlog-script:output}
hash-files = ${:command-line} hash-files = ${:command-line}
{% if slapparameter_dict.get('xlog_fluentbit_forward_host') %} {% if slapparameter_dict.get('xlog_fluentbit_forward_host') %}
[xlog-fluentbit-tag]
recipe = slapos.recipe.build
computer = ${slap-connection:computer-id}
enb-id = {{ slapparameter_dict.get("enb_id", "") }}
gnb-id = {{ slapparameter_dict.get("gnb_id", "") }}
init =
import socket
options['hostname'] = socket.gethostname()
radio_id = ''
if options['enb-id']:
radio_id = 'e%s' % options['enb-id']
elif options['gnb-id']:
radio_id = 'g%s' % options['gnb-id']
options['radio-id'] = radio_id
xlog_fluentbit_tag = '_'.join(options[x] for x in ('hostname', 'computer', 'radio-id') if options[x])
options['xlog-fluentbit-tag'] = xlog_fluentbit_tag
[xlog-fluentbit-config] [xlog-fluentbit-config]
recipe = slapos.recipe.template recipe = slapos.recipe.template
output = ${directory:etc}/${:_buildout_section_name_}.cfg output = ${directory:etc}/${:_buildout_section_name_}.cfg
...@@ -163,7 +184,9 @@ inline = ...@@ -163,7 +184,9 @@ inline =
[INPUT] [INPUT]
name tail name tail
path ${:logfile} path ${:logfile}
tag ${xlog-fluentbit-tag:xlog-fluentbit-tag}
Read_from_Head True Read_from_Head True
db ${directory:xlog-fluentbit}/tail-state
[OUTPUT] [OUTPUT]
name forward name forward
match * match *
...@@ -240,6 +263,9 @@ ru-list = {{ dumps(rulib.iru_dict.keys() | sort) }} ...@@ -240,6 +263,9 @@ ru-list = {{ dumps(rulib.iru_dict.keys() | sort) }}
cell-list = {{ dumps(rulib.icell_dict.keys() | sort) }} cell-list = {{ dumps(rulib.icell_dict.keys() | sort) }}
peer-list = {{ dumps(ipeer_dict.keys() | sort) }} peer-list = {{ dumps(ipeer_dict.keys() | sort) }}
peer-cell-list = {{ dumps(ipeercell_dict.keys() | sort) }} peer-cell-list = {{ dumps(ipeercell_dict.keys() | sort) }}
{%- if slapparameter_dict.get('xlog_fluentbit_forward_host') %}
fluentbit-tag = ${xlog-fluentbit-tag:xlog-fluentbit-tag}
{%- endif %}
[monitor-instance-parameter] [monitor-instance-parameter]
......
...@@ -46,7 +46,6 @@ import-list = ...@@ -46,7 +46,6 @@ import-list =
rawfile slaplte.jinja2 ${slaplte.jinja2:target} rawfile slaplte.jinja2 ${slaplte.jinja2:target}
rawfile ru_libinstance.jinja2.cfg ${ru_libinstance.jinja2.cfg:target} rawfile ru_libinstance.jinja2.cfg ${ru_libinstance.jinja2.cfg:target}
rawfile ru_sdr_libinstance.jinja2.cfg ${ru_sdr_libinstance.jinja2.cfg:target} rawfile ru_sdr_libinstance.jinja2.cfg ${ru_sdr_libinstance.jinja2.cfg:target}
rawfile ru_lopcomm_libinstance.jinja2.cfg ${ru_lopcomm_libinstance.jinja2.cfg:target}
rawfile ru_sunwave_libinstance.jinja2.cfg ${ru_sunwave_libinstance.jinja2.cfg:target} rawfile ru_sunwave_libinstance.jinja2.cfg ${ru_sunwave_libinstance.jinja2.cfg:target}
# activate eggs and modules used in jinja2 templates # activate eggs and modules used in jinja2 templates
...@@ -158,16 +157,6 @@ extra-context = ...@@ -158,16 +157,6 @@ extra-context =
raw sib23_template ${sib23.jinja2.asn:target} raw sib23_template ${sib23.jinja2.asn:target}
raw ru_amarisoft_stats_template ${ru_amarisoft-stats.jinja2.py:target} raw ru_amarisoft_stats_template ${ru_amarisoft-stats.jinja2.py:target}
raw ru_amarisoft_rf_info_template ${ru_amarisoft-rf-info.jinja2.py:target} raw ru_amarisoft_rf_info_template ${ru_amarisoft-rf-info.jinja2.py:target}
raw ru_lopcomm_stats_template ${ru_lopcomm_stats.jinja2.py:target}
raw ru_lopcomm_config_template ${ru_lopcomm_config.jinja2.py:target}
raw ru_lopcomm_software_template ${ru_lopcomm_software.jinja2.py:target}
raw ru_lopcomm_reset_info_template ${ru_lopcomm_reset-info.jinja2.py:target}
raw ru_lopcomm_reset_template ${ru_lopcomm_reset.jinja2.py:target}
raw ru_lopcomm_CreateProcessingEle_template ${ru_lopcomm_CreateProcessingEle.jinja2.xml:target}
raw ru_lopcomm_cu_config_template ${ru_lopcomm_cu_config.jinja2.xml:target}
raw ru_lopcomm_cu_inactive_config_template ${ru_lopcomm_cu_config.jinja2.xml:target}
raw ru_lopcomm_firmware_path ${ru_lopcomm_firmware-dl:target}
raw ru_lopcomm_firmware_filename ${ru_lopcomm_firmware-dl:filename}
raw ru_tapsplit ${ru_tapsplit:target} raw ru_tapsplit ${ru_tapsplit:target}
raw netcapdo ${netcapdo:exe} raw netcapdo ${netcapdo:exe}
raw openssl_location ${openssl:location} raw openssl_location ${openssl:location}
...@@ -217,16 +206,6 @@ extra-context = ...@@ -217,16 +206,6 @@ extra-context =
raw ru_amarisoft_stats_template ${ru_amarisoft-stats.jinja2.py:target} raw ru_amarisoft_stats_template ${ru_amarisoft-stats.jinja2.py:target}
raw ru_amarisoft_rf_info_template ${ru_amarisoft-rf-info.jinja2.py:target} raw ru_amarisoft_rf_info_template ${ru_amarisoft-rf-info.jinja2.py:target}
raw ru_lopcomm_stats_template ${ru_lopcomm_stats.jinja2.py:target}
raw ru_lopcomm_config_template ${ru_lopcomm_config.jinja2.py:target}
raw ru_lopcomm_software_template ${ru_lopcomm_software.jinja2.py:target}
raw ru_lopcomm_reset_info_template ${ru_lopcomm_reset-info.jinja2.py:target}
raw ru_lopcomm_reset_template ${ru_lopcomm_reset.jinja2.py:target}
raw ru_lopcomm_CreateProcessingEle_template ${ru_lopcomm_CreateProcessingEle.jinja2.xml:target}
raw ru_lopcomm_cu_config_template ${ru_lopcomm_cu_config.jinja2.xml:target}
raw ru_lopcomm_cu_inactive_config_template ${ru_lopcomm_cu_config.jinja2.xml:target}
raw ru_lopcomm_firmware_path ${ru_lopcomm_firmware-dl:target}
raw ru_lopcomm_firmware_filename ${ru_lopcomm_firmware-dl:filename}
raw ru_tapsplit ${ru_tapsplit:target} raw ru_tapsplit ${ru_tapsplit:target}
raw netcapdo ${netcapdo:exe} raw netcapdo ${netcapdo:exe}
raw ru_dnsmasq_template ${ru_dnsmasq.jinja2.cfg:target} raw ru_dnsmasq_template ${ru_dnsmasq.jinja2.cfg:target}
......
...@@ -3,7 +3,6 @@ ...@@ -3,7 +3,6 @@
[buildout] [buildout]
extends = extends =
sdr/buildout.cfg sdr/buildout.cfg
lopcomm/buildout.cfg
sunwave/buildout.cfg sunwave/buildout.cfg
parts += parts +=
......
...@@ -6,9 +6,6 @@ ...@@ -6,9 +6,6 @@
{ {
"$ref": "sdr/input-schema.json" "$ref": "sdr/input-schema.json"
}, },
{
"$ref": "lopcomm/input-schema.json"
},
{ {
"$ref": "sunwave/input-schema.json" "$ref": "sunwave/input-schema.json"
} }
......
...@@ -53,10 +53,8 @@ config-stats-period = {{ slapparameter_dict.get("enb_stats_fetch_period", 60) }} ...@@ -53,10 +53,8 @@ config-stats-period = {{ slapparameter_dict.get("enb_stats_fetch_period", 60) }}
{%- set jcell_ru_ref = slaplte.jcell_ru_ref %} {%- set jcell_ru_ref = slaplte.jcell_ru_ref %}
{%- set ierror = slaplte.ierror %} {%- set ierror = slaplte.ierror %}
{%- import 'ru_sdr_libinstance.jinja2.cfg' as rudrv_sdr with context %} {%- import 'ru_sdr_libinstance.jinja2.cfg' as rudrv_sdr with context %}
{%- import 'ru_lopcomm_libinstance.jinja2.cfg' as rudrv_lopcomm with context %}
{%- import 'ru_sunwave_libinstance.jinja2.cfg' as rudrv_sunwave with context %} {%- import 'ru_sunwave_libinstance.jinja2.cfg' as rudrv_sunwave with context %}
{%- set rudrv_dict = namespace(sdr=rudrv_sdr, {%- set rudrv_dict = namespace(sdr=rudrv_sdr,
lopcomm=rudrv_lopcomm,
sunwave=rudrv_sunwave) %} sunwave=rudrv_sunwave) %}
{%- set rudrv_init = {} %} {%- set rudrv_init = {} %}
......
<config xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
<processing-elements xmlns="urn:o-ran:processing-element:1.0">
<transport-session-type>CPRI-INTERFACE</transport-session-type>
<ru-elements>
<name>PE0</name>
<transport-flow>
<interface-name>eth1</interface-name>
</transport-flow>
</ru-elements>
</processing-elements>
</config>
\ No newline at end of file
# ru/lopcomm/buildout.cfg provides software code for handling Lopcomm ORAN Radio Units.
[buildout]
parts +=
ru_lopcomm_ncclient_common.py
[ru_lopcomm_libinstance.jinja2.cfg]
<= download-base
[ru_lopcomm_config.jinja2.py]
<= download-base
[ru_lopcomm_reset-info.jinja2.py]
<= download-base
[ru_lopcomm_reset.jinja2.py]
<= download-base
[ru_lopcomm_stats.jinja2.py]
<= download-base
[ru_lopcomm_software.jinja2.py]
<= download-base
[ru_lopcomm_ncclient_common.py]
<= download-base
destination = ${buildout:directory}/ncclient_common.py
[ru_lopcomm_CreateProcessingEle.jinja2.xml]
<= download-base
[ru_lopcomm_cu_config.jinja2.xml]
<= download-base
[ru_lopcomm_firmware-dl]
recipe = slapos.recipe.build:download
url = https://lab.nexedi.com/nexedi/ors-utils/raw/master/lopcomm-firmware/${:filename}
filename = PR.PRM61C70V1005.005.tar.gz
md5sum = 62281d0be42feac94e843e1850ba6e09
#!{{ python_path }}
import time
import sys
sys.path.append({{ repr(buildout_directory_path) }})
from ncclient_common import LopcommNetconfClient
if __name__ == '__main__':
nc = LopcommNetconfClient(log_file="{{ log_file }}")
while True:
try:
nc.connect("{{ netaddr.IPAddress(vtap.gateway) }}", 830, "oranuser", "oranpassword")
nc.edit_config(["{{ CreateProcessingEle_template }}", "{{ cu_inactive_config_template }}", "{{ cu_config_template }}"])
break
except Exception as e:
nc.logger.debug('Got exception, waiting 10 seconds before reconnecting...')
nc.logger.debug(e)
time.sleep(10)
finally:
nc.close()
<xc:config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">
<user-plane-configuration xc:operation="replace" xmlns="urn:o-ran:uplane-conf-option8:1.0">
<!-- TX path: eaxcid → TxEndpoint
mod → static TxEndpoint → TxArray
TxCarrier
(static TxEndpoint, TxArray and their association are defined by RU itself)
-->
{%- set TxCarrier = 'TXA0CC00' %}
{%- for ant in range(ru.n_antenna_dl) %}
{%- set port = ant // 2 %}
{%- set chan = ant % 2 %}
{%- set txep = 'TXA0P%02dC%02d' % (port, chan) %}
<!-- TxAntenna{{ ant }} -->
<tx-endpoints>
<name>{{ txep }}</name>
<e-axcid>
<o-du-port-bitmask>61440</o-du-port-bitmask>
<band-sector-bitmask>3968</band-sector-bitmask>
<ccid-bitmask>112</ccid-bitmask>
<ru-port-bitmask>15</ru-port-bitmask>
<eaxc-id>{{ ant }}</eaxc-id>
</e-axcid>
</tx-endpoints>
<tx-links>
<name>{{ txep }}</name>
<processing-element>PE0</processing-element>
<tx-array-carrier>{{ TxCarrier }}</tx-array-carrier>
<tx-endpoint>{{ txep }}</tx-endpoint>
</tx-links>
{%- endfor %}
<!--
RX path: eaxcid ← RxEndpoint
(data ∪ prach)
demod ← static RxEndpoint ← RxArray
RxCarrier
(static RxEndpoint, RxArray and their association are defined by RU itself)
-->
{%- set RxCarrier = 'RXA0CC00' %}
{%- for ant in range(ru.n_antenna_ul) %}
{%- set port = ant // 2 %}
{%- set chan = ant % 2 %}
{%- set rxep = 'RXA0P%02dC%02d' % (port, chan) %}
{%- set prachep = 'PRACH0P%02dC%02d' % (port, chan) %}
<!-- RxAntenna{{ ant }} -->
<rx-endpoints>
<name>{{ rxep }}</name>
<e-axcid>
<o-du-port-bitmask>61440</o-du-port-bitmask>
<band-sector-bitmask>3968</band-sector-bitmask>
<ccid-bitmask>112</ccid-bitmask>
<ru-port-bitmask>15</ru-port-bitmask>
<eaxc-id>{{ ant }}</eaxc-id>
</e-axcid>
</rx-endpoints>
<rx-endpoints>
<name>{{ prachep }}</name>
<e-axcid>
<o-du-port-bitmask>61440</o-du-port-bitmask>
<band-sector-bitmask>3968</band-sector-bitmask>
<ccid-bitmask>112</ccid-bitmask>
<ru-port-bitmask>15</ru-port-bitmask>
<eaxc-id>{{ 16*chan + 8 + port }}</eaxc-id>
</e-axcid>
</rx-endpoints>
<rx-links>
<name>{{ rxep }}</name>
<processing-element>PE0</processing-element>
<rx-array-carrier>{{ RxCarrier }}</rx-array-carrier>
<rx-endpoint>{{ rxep }}</rx-endpoint>
</rx-links>
<rx-links>
<name>{{ prachep }}</name>
<processing-element>PE0</processing-element>
<rx-array-carrier>{{ RxCarrier }}</rx-array-carrier>
<rx-endpoint>{{ prachep }}</rx-endpoint>
</rx-links>
{%- endfor %}
<!-- TX/RX carriers -->
<!-- TODO support multiple cells over 1 RU -->
{%- if cell.cell_type == 'lte' %}
{%- set dl_arfcn = cell.dl_earfcn %}
{%- set ul_arfcn = cell.ul_earfcn %}
{%- set dl_freq = int(xearfcn_module.frequency(dl_arfcn) * 1e6) %}
{%- set ul_freq = int(xearfcn_module.frequency(ul_arfcn) * 1e6) %}
{%- elif cell.cell_type == 'nr' %}
{%- set dl_arfcn = cell.dl_nr_arfcn %}
{%- set ul_arfcn = cell.ul_nr_arfcn %}
{%- set dl_freq = int(xnrarfcn_module.frequency(dl_arfcn) * 1e6) %}
{%- set ul_freq = int(xnrarfcn_module.frequency(ul_arfcn) * 1e6) %}
{%- else %}
{%- do bug('unreachable') %}
{%- endif %}
{%- set bw = int(cell.bandwidth * 1e6) %}
<tx-array-carriers>
<name>{{ TxCarrier }}</name>
<absolute-frequency-center>{{ dl_arfcn }}</absolute-frequency-center>
<center-of-channel-bandwidth>{{ dl_freq }}</center-of-channel-bandwidth>
<channel-bandwidth>{{ bw }}</channel-bandwidth>
<active>{{ ru.txrx_active }}</active>
<rw-type>{{ cell.cell_type | upper }}</rw-type>
<rw-duplex-scheme>{{ cell.rf_mode | upper }}</rw-duplex-scheme>
<gain>{{ ru.tx_gain }}</gain>
<downlink-radio-frame-offset>0</downlink-radio-frame-offset>
<downlink-sfn-offset>0</downlink-sfn-offset>
</tx-array-carriers>
<rx-array-carriers>
<name>{{ RxCarrier }}</name>
<absolute-frequency-center>{{ ul_arfcn }}</absolute-frequency-center>
<center-of-channel-bandwidth>{{ ul_freq }}</center-of-channel-bandwidth>
<channel-bandwidth>{{ bw }}</channel-bandwidth>
<active>{{ ru.txrx_active }}</active>
<downlink-radio-frame-offset>0</downlink-radio-frame-offset>
<downlink-sfn-offset>0</downlink-sfn-offset>
<!-- <gain>{{ ru.rx_gain }}</gain> -->
<!-- TODO(lu.xu): clarify with Lopcomm regaring rx gain -->
<gain-correction>0.0</gain-correction>
<n-ta-offset>0</n-ta-offset>
</rx-array-carriers>
</user-plane-configuration>
</xc:config>
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "Lopcomm ORAN",
"type": "object",
"required": [
"ru_type",
"ru_link_type",
"n_antenna_dl",
"n_antenna_ul",
"tx_gain",
"rx_gain",
"cpri_link",
"mac_addr"
],
"properties": {
"$ref": "../../ru/common.json#/properties",
"ru_type": {
"$ref": "#/properties/ru_type",
"const": "lopcomm"
},
"ru_link_type": {
"$ref": "#/properties/ru_link_type",
"const": "cpri"
},
"n_antenna_dl": {
"$ref": "#/properties/n_antenna_dl",
"default": 2
},
"n_antenna_ul": {
"$ref": "#/properties/n_antenna_ul",
"default": 2
},
"cpri_link": {
"$ref": "#/properties/cpri_link",
"properties": {
"$ref": "#/properties/cpri_link/properties",
"mapping": {
"$ref": "#/properties/cpri_link/properties/mapping",
"const": "hw",
"enum": [
"hw"
]
},
"rx_delay": {
"$ref": "#/properties/cpri_link/properties/rx_delay",
"default": 25.11
},
"tx_delay": {
"$ref": "#/properties/cpri_link/properties/tx_delay",
"default": 14.71
},
"tx_dbm": {
"$ref": "#/properties/cpri_link/properties/tx_dbm",
"default": 63
}
}
},
"reset_schedule": {
"title": "Cron schedule for RRH reset",
"description": "Refer https://crontab.guru/ to make a reset schedule for RRH, for example, '0 1 * * *' means the RRH will reset every day at 1 am",
"type": "string"
}
}
}
{#- Package ru/lopcomm/libinstance provides instance code for handling Lopcomm ORAN Radio Units. #}
{%- macro buildout_iru(iru, icell_list) %}
{%- set ru_ref = J(jref_of_shared(iru)) %}
{%- set ru = iru['_'] %}
{%- set ns = namespace(inactive_ru=ru.copy()) %}
{%- do ns.inactive_ru.update({'txrx_active': 'INACTIVE'}) %}
{%- if len(icell_list) != 1 %}
{%- do ierror(iru, 'ru/lopcomm supports only 1 cell ; requested %d' % len(icell_list)) %}
{%- endif %}
{%- set icell = icell_list[0] %}
{%- set cell = icell['_'] %}
{#- indicate whether RU is listening for netconf #}
{%- if not testing %}
{{ promise('%s-netconf-socket' % ru_ref) }}
promise = check_socket_listening
config-host = ${vtap.{{ru.cpri_link._tap}}:gateway}
config-port = 830
{%- endif %}
{#- push firmware to RU #}
{{ part('%s-software-template' % ru_ref) }}
recipe = slapos.recipe.template:jinja2
extensions = jinja2.ext.do
_logbase = ${directory:var}/log/{{B('%s-software' % ru_ref)}}
log-output = ${:_logbase}.log
software-reply-json-log-output = ${:_logbase}-reply.json.log
remote-file-path = sftp://${user-info:pw-name}@[${sshd-service:ipv6}]:${sshd-service:port}{{ru_lopcomm_firmware_path}}
is_firmware_updated = ${directory:etc}/{{B('%s.is_firmware_updated' % ru_ref)}}
context =
section directory directory
section vtap vtap.{{ ru.cpri_link._tap }}
key slapparameter_dict myslap:parameter_dict
key log_file :log-output
key software_reply_json_log_file :software-reply-json-log-output
key remote_file_path :remote-file-path
raw testing {{ testing }}
raw python_path {{ buildout_directory}}/bin/pythonwitheggs
raw buildout_directory_path {{ buildout_directory }}
key is_firmware_updated :is_firmware_updated
raw firmware_name {{ru_lopcomm_firmware_filename}}
import netaddr netaddr
mode = 0775
url = {{ ru_lopcomm_software_template }}
output = ${directory:script}/{{B('%s-software.py' % ru_ref)}}
{%- if not testing %}
{{ promise('%s-firmware' % ru_ref) }}
promise = check_command_execute
config-command = [ -f ${ {{-B('%s-software-template' % ru_ref)}}:is_firmware_updated} ]
{%- endif %}
[{{ B('%s-cu-config' % ru_ref) }}]
<= config-base
url = {{ ru_lopcomm_cu_config_template }}
output = ${directory:etc}/{{B('%s-cu_config.xml' % ru_ref)}}
extra-context =
import xearfcn_module xlte.earfcn
import xnrarfcn_module xlte.nrarfcn
key ru :ru
key cell :cell
ru = {{ dumps(ru) }}
cell = {{ dumps(cell) }}
[{{ B('%s-cu-inactive-config' % ru_ref) }}]
<= config-base
url = {{ ru_lopcomm_cu_config_template }}
output = ${directory:etc}/{{B('%s-cu_inactive_config.xml' % ru_ref)}}
extra-context =
import xearfcn_module xlte.earfcn
import xnrarfcn_module xlte.nrarfcn
key ru :ru
key cell :cell
ru = {{ dumps(ns.inactive_ru) }}
cell = {{ dumps(cell) }}
[{{ B('%s-config-template' % ru_ref) }}]
recipe = slapos.recipe.template:jinja2
extensions = jinja2.ext.do
log-output = ${directory:var}/log/{{B('%s-config.log' % ru_ref)}}
context =
section directory directory
section vtap vtap.{{ ru.cpri_link._tap }}
key log_file :log-output
raw testing {{ testing }}
raw python_path {{ buildout_directory}}/bin/pythonwitheggs
raw buildout_directory_path {{ buildout_directory }}
raw CreateProcessingEle_template {{ ru_lopcomm_CreateProcessingEle_template }}
key cu_config_template {{B('%s-cu-config' % ru_ref)}}:output
key cu_inactive_config_template {{B('%s-cu-inactive-config' % ru_ref)}}:output
import netaddr netaddr
mode = 0775
url = {{ ru_lopcomm_config_template }}
output = ${directory:script}/{{B('%s-config.py' % ru_ref)}}
{{ promise('%s-config-log' % ru_ref) }}
promise = check_lopcomm_config_log
config-config-log = ${ {{-B('%s-config-template' % ru_ref)}}:log-output}
{#- handle notifications from RU + keep on touching RU watchdog #}
[{{ B('%s-stats-template' % ru_ref) }}]
recipe = slapos.recipe.template:jinja2
extensions = jinja2.ext.do
_logbase = ${directory:var}/log/{{B('%s' % ru_ref)}}
log-output = ${:_logbase}-stats.log
json-log-output = ${:_logbase}-stats.json.log
cfg-json-log-output = ${:_logbase}-config.json.log
supervision-json-log-output = ${:_logbase}-supervision.json.log
ncsession-json-log-output = ${:_logbase}-ncsession.json.log
software-json-log-output = ${:_logbase}-software.json.log
supervision-reply-json-log-output = ${:_logbase}-supervision-reply.json.log
is_netconf_connected = ${directory:etc}/{{B('%s.is_netconf_connected' % ru_ref)}}
context =
section directory directory
section vtap vtap.{{ ru.cpri_link._tap }}
key slapparameter_dict myslap:parameter_dict
key log_file :log-output
key json_log_file :json-log-output
key cfg_json_log_file :cfg-json-log-output
key supervision_json_log_file :supervision-json-log-output
key supervision_reply_json_log_file :supervision-reply-json-log-output
key is_netconf_connected :is_netconf_connected
key ncsession_json_log_file :ncsession-json-log-output
key software_json_log_file :software-json-log-output
raw testing {{ testing }}
raw python_path {{ buildout_directory}}/bin/pythonwitheggs
raw buildout_directory_path {{ buildout_directory }}
import netaddr netaddr
mode = 0775
url = {{ ru_lopcomm_stats_template }}
output = ${directory:bin}/{{B('%s-stats.py' % ru_ref)}}
{{ part('%s-stats-service' % ru_ref) }}
recipe = slapos.cookbook:wrapper
command-line = ${ {{-B('%s-stats-template' % ru_ref)}}:output}
wrapper-path = ${directory:service}/{{B('%s-stats' % ru_ref)}}
mode = 0775
hash-files =
${:command-line}
{%- if not testing %}
{{ promise('%s-netconf-connection' % ru_ref) }}
promise = check_command_execute
config-command = [ -f ${ {{-B('%s-stats-template' % ru_ref)}}:is_netconf_connected} ]
{%- endif %}
{{ promise('%s-vswr' % ru_ref) }}
promise = check_lopcomm_vswr
config-netconf-log = ${ {{-B('%s-stats-template' % ru_ref)}}:json-log-output}
{{ promise('%s-rssi' % ru_ref) }}
promise = check_lopcomm_rssi
config-netconf-log = ${ {{-B('%s-stats-template' % ru_ref)}}:json-log-output}
{{ promise('%s-pa-current' % ru_ref) }}
promise = check_lopcomm_pa_current
config-netconf-log = ${ {{-B('%s-stats-template' % ru_ref)}}:json-log-output}
{{ promise('%s-pa-output-power' % ru_ref) }}
promise = check_lopcomm_pa_output_power
config-netconf-log = ${ {{-B('%s-stats-template' % ru_ref)}}:json-log-output}
{{ promise('%s-sync' % ru_ref) }}
promise = check_lopcomm_sync
config-netconf-log = ${ {{-B('%s-stats-template' % ru_ref)}}:json-log-output}
{{ promise('%s-lof' % ru_ref) }}
promise = check_lopcomm_lof
config-netconf-log = ${ {{-B('%s-stats-template' % ru_ref)}}:json-log-output}
{{ promise('%s-stats-log' % ru_ref) }}
promise = check_lopcomm_stats_log
config-stats-log = ${ {{-B('%s-stats-template' % ru_ref)}}:log-output}
{#- reset RU periodically #}
{%- if ru.get("reset_schedule") %}
[{{ B('%s-reset-info-template' % ru_ref) }}]
recipe = slapos.recipe.template:jinja2
extensions = jinja2.ext.do
_logbase = ${directory:var}/log/{{B('%s-reset-info' % ru_ref)}}
log-output = ${:_logbase}.log
json-log-output = ${:_logbase}.json.log
context =
section vtap vtap.{{ ru.cpri_link._tap }}
key log_file :log-output
key json_log_file :json-log-output
raw stats_period {{ slapparameter_dict.get("enb_stats_fetch_period", 60) }}
raw testing {{ testing }}
raw python_path {{ buildout_directory}}/bin/pythonwitheggs
import netaddr netaddr
mode = 0775
url = {{ ru_lopcomm_reset_info_template }}
output = ${directory:bin}/{{B('%s-reset-info.py' % ru_ref)}}
[{{ B('%s-reset-template' % ru_ref) }}]
recipe = slapos.recipe.template:jinja2
extensions = jinja2.ext.do
_logbase = ${directory:var}/log/{{B('%s-reset' % ru_ref)}}
log-output = ${:_logbase}.log
json-log-output = ${:_logbase}.json.log
context =
section vtap vtap.{{ ru.cpri_link._tap }}
key log_file :log-output
raw python_path {{ buildout_directory}}/bin/pythonwitheggs
raw buildout_directory_path {{ buildout_directory }}
import netaddr netaddr
mode = 0775
url = {{ ru_lopcomm_reset_template }}
output = ${directory:etc}/{{B('%s-reset.py' % ru_ref)}}
{{ part('%s-reset-cron' % ru_ref) }}
recipe = slapos.cookbook:cron.d
cron-entries = ${cron:cron-entries}
name = {{B('%s-reset' % ru_ref)}}
frequency = {{ ru.reset_schedule }}
command = {{ buildout_directory}}/bin/pythonwitheggs ${ {{-B('%s-reset-template' % ru_ref)}}:output}
{{ part('%s-reset-info-service' % ru_ref) }}
recipe = slapos.cookbook:wrapper
command-line = ${ {{-B('%s-reset-info-template' % ru_ref)}}:output}
wrapper-path = ${directory:service}/{{B('%s-reset-info' % ru_ref)}}
mode = 0775
hash-files =
${:command-line}
{%- endif %}
{#- amend RU-published information with Lopcomm-specific bits #}
[{{ B('ipublish-%s' % ru_ref) }}]
bbu-ssh-command = ssh ${user-info:pw-name}@${sshd-service:ipv6} -p ${sshd-service:port}
bbu-ssh-url = ssh://${user-info:pw-name}@[${sshd-service:ipv6}]:${sshd-service:port}
firmware = {{ru_lopcomm_firmware_filename}}
{%- endmacro %}
{%- macro buildout() %}
# deploy openssh-server for software upgrade
#
# FIXME user-authorized-key is global for eNB. Either we need to put SSH server
# to be also global, or unroll an SSH server via paramiko inside
# ru/lopcomm/software.py just to handle software upgrades there.
[user-info]
recipe = slapos.cookbook:userinfo
[sshd-port]
recipe = slapos.cookbook:free_port
minimum = 22222
maximum = 22231
ip = {{my_ipv6}}
[sshd-config]
recipe = slapos.recipe.template:jinja2
output = ${directory:etc}/sshd.conf
path_pid = ${directory:run}/sshd.pid
inline =
PidFile ${:path_pid}
Port ${sshd-port:port}
ListenAddress ${sshd-port:ip}
Protocol 2
HostKey ${sshd-ssh-host-rsa-key:output}
HostKey ${sshd-ssh-host-ecdsa-key:output}
PasswordAuthentication no
PubkeyAuthentication yes
HostKeyAlgorithms ssh-rsa,rsa-sha2-512,rsa-sha2-256,ecdsa-sha2-nistp521
AuthorizedKeysFile ${buildout:directory}/.ssh/authorized_keys
Subsystem sftp {{ openssh_location }}/libexec/sftp-server
{{ part('sshd-service') }}
recipe = slapos.cookbook:wrapper
command-line = {{ openssh_location }}/sbin/sshd -D -e -f ${sshd-config:output}
wrapper-path = ${directory:service}/sshd
hash-files = ${sshd-config:output}
environment =
HOME=${directory:home}
ipv6 = ${sshd-port:ip}
port = ${sshd-port:port}
{{ part('sshd-add-authorized-key') }}
recipe = slapos.cookbook:dropbear.add_authorized_key
home = ${buildout:directory}
key = {{ slapparameter_dict.get("user-authorized-key", '') }}
[sshd-ssh-keygen-base]
recipe = plone.recipe.command
output = ${directory:etc}/${:_buildout_section_name_}
command = {{ openssh_output_keygen }} -f ${:output} -N '' ${:extra-args}
[sshd-ssh-host-rsa-key]
<=sshd-ssh-keygen-base
extra-args=-t rsa
[sshd-ssh-host-ecdsa-key]
<=sshd-ssh-keygen-base
extra-args=-t ecdsa -b 521
{{ promise('sshd') }}
promise = check_socket_listening
config-host = ${sshd-service:ipv6}
config-port = ${sshd-service:port}
{%- endmacro %}
import time
import logging
import json
import xmltodict
from logging.handlers import RotatingFileHandler
from ncclient import manager
from ncclient.operations import RPCError
from ncclient.xml_ import *
from ncclient.devices.default import DefaultDeviceHandler
class LopcommNetconfClient:
def __init__(self, log_file, json_log_file=None, cfg_json_log_file=None, supervision_json_log_file=None, ncsession_json_log_file=None, software_json_log_file=None, software_reply_json_log_file=None, supervision_reply_json_log_file=None, testing=False):
self.logger = logging.getLogger('logger')
self.logger.setLevel(logging.DEBUG)
handler = RotatingFileHandler(log_file, maxBytes=100000, backupCount=5)
self.logger.addHandler(handler)
formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
handler.setFormatter(formatter)
if json_log_file:
self.json_logger = logging.getLogger('json_logger')
self.json_logger.setLevel(logging.DEBUG)
json_handler = RotatingFileHandler(json_log_file, maxBytes=100000, backupCount=5)
json_formatter = logging.Formatter('{"time": "%(asctime)s", "log_level": "%(levelname)s", "message": "%(message)s", "data": %(data)s}')
json_handler.setFormatter(json_formatter)
self.json_logger.addHandler(json_handler)
self.cfg_json_logger = logging.getLogger('cfg_json_logger')
self.cfg_json_logger.setLevel(logging.DEBUG)
cfg_json_handler = RotatingFileHandler(cfg_json_log_file, maxBytes=100000, backupCount=5)
cfg_json_formatter = logging.Formatter('{"time": "%(asctime)s", "log_level": "%(levelname)s", "message": "%(message)s", "data": %(data)s}')
cfg_json_handler.setFormatter(cfg_json_formatter)
self.cfg_json_logger.addHandler(cfg_json_handler)
self.supervision_json_logger = logging.getLogger('supervision_json_logger')
self.supervision_json_logger.setLevel(logging.DEBUG)
supervision_json_handler = RotatingFileHandler(supervision_json_log_file, maxBytes=100000, backupCount=5)
supervision_json_formatter = logging.Formatter('{"time": "%(asctime)s", "log_level": "%(levelname)s", "message": "%(message)s", "data": %(data)s}')
supervision_json_handler.setFormatter(supervision_json_formatter)
self.supervision_json_logger.addHandler(supervision_json_handler)
self.ncsession_json_logger = logging.getLogger('ncsession_json_logger')
self.ncsession_json_logger.setLevel(logging.DEBUG)
ncsession_json_handler = RotatingFileHandler(ncsession_json_log_file, maxBytes=100000, backupCount=5)
ncsession_json_formatter = logging.Formatter('{"time": "%(asctime)s", "log_level": "%(levelname)s", "message": "%(message)s", "data": %(data)s}')
ncsession_json_handler.setFormatter(ncsession_json_formatter)
self.ncsession_json_logger.addHandler(ncsession_json_handler)
self.software_json_logger = logging.getLogger('software_json_logger')
self.software_json_logger.setLevel(logging.DEBUG)
software_json_handler = RotatingFileHandler(software_json_log_file, maxBytes=100000, backupCount=5)
software_json_formatter = logging.Formatter('{"time": "%(asctime)s", "log_level": "%(levelname)s", "message": "%(message)s", "data": %(data)s}')
software_json_handler.setFormatter(software_json_formatter)
self.software_json_logger.addHandler(software_json_handler)
else:
self.json_logger = None
self.cfg_json_logger = None
self.supervision_json_logger = None
self.ncsession_json_logger = None
self.software_json_logger = None
if supervision_reply_json_log_file:
self.supervision_reply_json_logger = logging.getLogger('supervision_reply_json_logger')
self.supervision_reply_json_logger.setLevel(logging.DEBUG)
supervision_reply_json_handler = RotatingFileHandler(supervision_reply_json_log_file, maxBytes=100000, backupCount=5)
supervision_reply_json_formatter = logging.Formatter('{"time": "%(asctime)s", "log_level": "%(levelname)s", "message": "%(message)s", "data": %(data)s}')
supervision_reply_json_handler.setFormatter(supervision_reply_json_formatter)
self.supervision_reply_json_logger.addHandler(supervision_reply_json_handler)
else:
self.supervision_reply_json_logger = None
if software_reply_json_log_file:
self.software_reply_json_logger = logging.getLogger('software_reply_json_logger')
self.software_reply_json_logger.setLevel(logging.DEBUG)
software_reply_json_handler = RotatingFileHandler(software_reply_json_log_file, maxBytes=100000, backupCount=5)
software_reply_json_formatter = logging.Formatter('{"time": "%(asctime)s", "log_level": "%(levelname)s", "message": "%(message)s", "data": %(data)s}')
software_reply_json_handler.setFormatter(software_reply_json_formatter)
self.software_reply_json_logger.addHandler(software_reply_json_handler)
else:
self.software_reply_json_logger = None
if testing:
return
def connect(self, host, port, user, password):
self.address = (host, port)
self.logger.info('Connecting to %s, user %s...' % (self.address, user))
self.conn = manager.connect(host=host,
port=port,
username=user,
password=password,
timeout=1800,
device_params={
'name': 'default'
},
hostkey_verify=False)
self.logger.info('Connection to %s successful' % (self.address,))
def subscribe(self):
sub = self.conn.create_subscription()
self.logger.info('Subscription to %s successful' % (self.address,))
def get_notification(self):
self.logger.debug('Waiting for notification from %s...' % (self.address,))
result = self.conn.take_notification(block=True, timeout=120)
if result:
self.logger.debug('Got new notification from %s...' % (self.address,))
result_in_xml = result._raw
data_dict = xmltodict.parse(result_in_xml)
if 'alarm-notif' in data_dict['notification']:
self.json_logger.info('', extra={'data': json.dumps(data_dict)})
elif 'supervision-notification' in data_dict['notification']:
self.supervision_json_logger.info('', extra={'data': json.dumps(data_dict)})
elif 'netconf-session-start' in data_dict['notification'] or 'netconf-session-end' in data_dict['notification']:
self.ncsession_json_logger.info('', extra={'data': json.dumps(data_dict)})
elif any(event in data_dict['notification'] for event in ['install-event', 'activation-event', 'download-event']):
self.software_json_logger.info('', extra={'data': json.dumps(data_dict)})
else:
self.cfg_json_logger.info('', extra={'data': json.dumps(data_dict)})
else:
raise TimeoutError
def edit_config(self, config_files):
for config_file in config_files:
with open(config_file) as f:
config_xml = f.read()
try:
self.logger.info('Sending edit-config RPC request...')
self.conn.edit_config(target='running', config=config_xml)
self.logger.info('Edit-config RPC request sent successfully')
except RPCError as e:
self.logger.error('Error sending edit-config RPC request: %s' % e)
def custom_rpc_request(self, rpc_xml):
try:
self.logger.info('Sending custom RPC request...')
response = self.conn.dispatch(to_ele(rpc_xml))
if response.ok:
self.logger.info('Custom RPC request sent successfully')
return response.xml
else:
self.logger.error('Error sending custom RPC request: %s' % response.error)
except RPCError as e:
self.logger.error('Error sending custom RPC request: %s' % e)
def reset_device(self):
self.logger.info('Resetting...')
reset_rpc_xml = """
<reset xmlns="urn:o-ran:operations:1.0">
</reset>
"""
reset_reply_xml = self.custom_rpc_request(reset_rpc_xml)
if reset_reply_xml:
reset_data = xmltodict.parse(reset_reply_xml)
if self.software_reply_json_logger:
self.software_reply_json_logger.info('', extra={'data': json.dumps(reset_data)})
self.logger.info('Wait 60 second then reboot!')
time.sleep(60)
def get_inventory(self):
self.logger.info('Fetching software inventory...')
inventory_rpc_xml = """
<get xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
<filter type="subtree">
<software-inventory xmlns="urn:o-ran:software-management:1.0" />
</filter>
</get>
"""
inventory_reply_xml = self.custom_rpc_request(inventory_rpc_xml)
if inventory_reply_xml:
self.logger.info('Finish fetching software inventory.')
inventory_data = xmltodict.parse(inventory_reply_xml)
self.software_reply_json_logger.info('', extra={'data': json.dumps(inventory_data)})
nonrunning_slot_name = None
running_slot_name = None
active_nonrunning_slot_name = None
nonrunning_slot_name_build_version = None
running_slot_name_build_version = None
software_slots = inventory_data['nc:rpc-reply']['data']['software-inventory']['software-slot']
for slot in software_slots:
if slot['running'] == 'false':
nonrunning_slot_name = slot['name']
nonrunning_slot_name_build_version = slot['build-version']
if slot['running'] == 'true':
running_slot_name = slot['name']
running_slot_name_build_version = slot['build-version']
elif slot['active'] == 'true' and slot['running'] == 'false':
active_nonrunning_slot_name = slot['name']
return {
"nonrunning_slot_name": nonrunning_slot_name,
"running_slot_name": running_slot_name,
"active_nonrunning_slot_name": active_nonrunning_slot_name,
"nonrunning_slot_name_build_version": nonrunning_slot_name_build_version,
"running_slot_name_build_version": running_slot_name_build_version
}
def supervision_reset(self, interval=60, margin=10):
self.logger.info("NETCONF server supervision replying...")
supervision_watchdog_rpc_xml = f"""
<supervision-watchdog-reset xmlns="urn:o-ran:supervision:1.0">
<supervision-notification-interval>{interval}</supervision-notification-interval>
<guard-timer-overhead>{margin}</guard-timer-overhead>
</supervision-watchdog-reset>
"""
supervision_watchdog_reply_xml = self.custom_rpc_request(supervision_watchdog_rpc_xml)
replied = False
if supervision_watchdog_reply_xml:
replied = True
self.logger.info("NETCONF server supervision replied")
supervision_watchdog_data = xmltodict.parse(supervision_watchdog_reply_xml)
self.supervision_reply_json_logger.info('', extra={'data': json.dumps(supervision_watchdog_data)})
return replied
def close(self):
# Close not compatible between ncclient and netconf server
#self.conn.close()
pass
#!{{ python_path }}
import paramiko
import logging
from logging.handlers import RotatingFileHandler
def get_uptime(hostname, username, password):
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
client.connect(hostname, username=username, password=password)
stdin, stdout, stderr = client.exec_command('uptime')
uptime_output = stdout.read().decode()
return uptime_output
except Exception as e:
logger.info(f"Error: {e}")
finally:
client.close()
# Usage
hostname = "{{ netaddr.IPAddress(vtap.gateway) }}"
username = "oranuser"
password = "oranpassword"
# Initialize logger
log_file = "{{ log_file }}"
logger = logging.getLogger('logger')
logger.setLevel(logging.INFO)
handler = RotatingFileHandler(log_file, maxBytes=30000, backupCount=2)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
if {{ testing }}:
pass
else:
rrh_uptime = get_uptime(hostname, username, password)
logger.info(f"Uptime from RRH: {rrh_uptime}")
#!{{ python_path }}
import time
import sys
sys.path.append({{ repr(buildout_directory_path) }})
from ncclient_common import LopcommNetconfClient
if __name__ == '__main__':
nc = LopcommNetconfClient(log_file="{{ log_file }}")
try:
nc.connect("{{ netaddr.IPAddress(vtap.gateway) }}", 830, "oranuser", "oranpassword")
nc.reset_device()
nc.logger.info("Device reset successful.")
except Exception as e:
nc.logger.debug('Got exception while resetting...')
nc.logger.debug(e)
finally:
nc.close()
#!{{ python_path }}
import time
import json
import xmltodict
import sys
import re
import os
sys.path.append({{ repr(buildout_directory_path) }})
from ncclient_common import LopcommNetconfClient
if __name__ == '__main__':
nc = LopcommNetconfClient(
log_file="{{ log_file }}",
software_reply_json_log_file="{{ software_reply_json_log_file }}"
)
while True:
try:
firmware_check_file= '{{ is_firmware_updated }}'
nc.connect("{{ netaddr.IPAddress(vtap.gateway) }}", 830, "oranuser", "oranpassword")
# Fetch software inventory
inventory_vars = nc.get_inventory()
nonrunning_slot_name = inventory_vars["nonrunning_slot_name"]
running_slot_name = inventory_vars["running_slot_name"]
active_nonrunning_slot_name = inventory_vars["active_nonrunning_slot_name"]
nonrunning_slot_name_build_version = inventory_vars["nonrunning_slot_name_build_version"]
running_slot_name_build_version = inventory_vars["running_slot_name_build_version"]
if running_slot_name and nonrunning_slot_name:
if running_slot_name:
nc.logger.info("One slot is running and one is non-running. Proceeding...")
if running_slot_name_build_version in "{{firmware_name}}":
if not os.path.exists(firmware_check_file):
open(firmware_check_file, "w").write('True')
nc.logger.info("Running slot's build-version %s is already updated. Skipping install." % running_slot_name_build_version)
else:
if os.path.exists(firmware_check_file):
os.remove(firmware_check_file)
nc.logger.info("Current build version: %s" % running_slot_name_build_version)
user_authorized_key ="""{{ slapparameter_dict.get('user-authorized-key', '') }}"""
match = re.match(r'ssh-rsa ([^\s]+)', user_authorized_key)
if match:
extracted_key = match.group(1)
else:
nc.logger.info("No valid key found in user authorized key.")
download_rpc_xml = f"""
<software-download xmlns="urn:o-ran:software-management:1.0">
<remote-file-path>{{remote_file_path}}</remote-file-path>
<server>
<keys>
<algorithm xmlns:ct="urn:ietf:params:xml:ns:yang:ietf-crypto-types">1024</algorithm>
<public-key>{extracted_key}</public-key>
</keys>
</server>
</software-download>
"""
download_reply_xml = nc.custom_rpc_request(download_rpc_xml)
nc.logger.info("Downloading software...")
time.sleep(60)
if download_reply_xml:
nc.logger.info("Download proceed.")
download_data = xmltodict.parse(download_reply_xml)
nc.software_reply_json_logger.info('', extra={'data': json.dumps(download_data)})
install_rpc_xml = f"""
<software-install xmlns="urn:o-ran:software-management:1.0">
<slot-name>{nonrunning_slot_name}</slot-name>
<file-names>{{firmware_name}}</file-names>
</software-install>
"""
install_reply_xml = nc.custom_rpc_request(install_rpc_xml)
nc.logger.info("Installing software...")
time.sleep(60)
if install_reply_xml:
nc.logger.info("Installation proceed.")
install_data = xmltodict.parse(install_reply_xml)
nc.software_reply_json_logger.info('', extra={'data': json.dumps(install_data)})
if nonrunning_slot_name_build_version in "{{firmware_name}}":
activate_rpc_xml = f"""
<software-activate xmlns="urn:o-ran:software-management:1.0">
<slot-name>{nonrunning_slot_name}</slot-name>
</software-activate>
"""
activate_reply_xml = nc.custom_rpc_request(activate_rpc_xml)
nc.logger.info("Activating software...")
time.sleep(60)
if activate_reply_xml:
nc.logger.info("Activation proceed.")
activate_data = xmltodict.parse(activate_reply_xml)
nc.software_reply_json_logger.info('', extra={'data': json.dumps(activate_data)})
nc.get_inventory()
if nonrunning_slot_name_build_version in "{{firmware_name}}" and active_nonrunning_slot_name:
nc.logger.info("Active non-running slot has the updated build version. Resetting device.")
nc.reset_device()
break
except Exception as e:
nc.logger.debug('Got exception, waiting 10 seconds before reconnecting...')
nc.logger.debug(str(e))
time.sleep(10)
finally:
nc.close()
#!{{ python_path }}
import time
import sys
import os
import threading
sys.path.append({{ repr(buildout_directory_path) }})
from ncclient_common import LopcommNetconfClient
# Shared variable to indicate error occurred
error_occurred = False
lock = threading.Lock()
def get_notification_continuously(nc):
global error_occurred
try:
while not error_occurred:
nc.get_notification()
pass
except Exception as e:
with lock:
error_occurred = True
nc.logger.error(f'Error in get_notification_continuously: {e}')
# supervision watchdog keeps on
def run_supervision_reset_continuously(nc):
global error_occurred
netconf_check_file = '{{ is_netconf_connected }}'
interval = 60
margin = 10
try:
while not error_occurred:
t0 = time.time()
replied = nc.supervision_reset(interval, margin)
if replied:
with open(netconf_check_file, "w") as f:
f.write('')
elif os.path.exists(netconf_check_file):
os.remove(netconf_check_file)
t1 = time.time()
time.sleep(interval - (t1 - t0))
except Exception as e:
with lock:
error_occurred = True
nc.logger.error(f'Error in run_supervision_reset_continuously: {e}')
if __name__ == '__main__':
nc = LopcommNetconfClient(
log_file="{{ log_file }}",
json_log_file="{{ json_log_file }}",
cfg_json_log_file="{{ cfg_json_log_file }}",
supervision_json_log_file="{{ supervision_json_log_file }}",
ncsession_json_log_file="{{ ncsession_json_log_file }}",
software_json_log_file="{{ software_json_log_file }}",
supervision_reply_json_log_file="{{ supervision_reply_json_log_file }}"
)
threads = []
try:
nc.connect("{{ netaddr.IPAddress(vtap.gateway) }}", 830, "oranuser", "oranpassword")
nc.subscribe()
notification_thread = threading.Thread(target=get_notification_continuously, args=(nc,))
supervision_thread = threading.Thread(target=run_supervision_reset_continuously, args=(nc,))
threads.append(notification_thread)
threads.append(supervision_thread)
for thread in threads:
thread.start()
for thread in threads:
thread.join()
except Exception as e:
nc.logger.debug('Got exception, waiting 10 seconds before reconnecting...')
nc.logger.debug(e)
time.sleep(10)
finally:
nc.close()
...@@ -34,17 +34,6 @@ ...@@ -34,17 +34,6 @@
'tx_dbm': 0, 'tx_dbm': 0,
}, },
'ru/lopcomm': {
'n_antenna_dl': 2,
'n_antenna_ul': 2,
},
'ru/lopcomm/cpri_link': {
'mapping': 'hw',
'rx_delay': 25.11,
'tx_delay': 14.71,
'tx_dbm': 63,
},
'ru/sunwave': { 'ru/sunwave': {
'n_antenna_dl': 2, 'n_antenna_dl': 2,
'n_antenna_ul': 1, 'n_antenna_ul': 1,
......
...@@ -502,171 +502,6 @@ class SDR4: ...@@ -502,171 +502,6 @@ class SDR4:
)) ))
# Lopcomm4 is mixin to verify Lopcomm driver wrt all LTE/NR x FDD/TDD modes.
class Lopcomm4:
@classmethod
def RUcfg(cls, i):
return {
'ru_type': 'lopcomm',
'ru_link_type': 'cpri',
'cpri_link': {
'sdr_dev': 0,
'sfp_port': i,
'mult': 4,
'mapping': 'hw',
'rx_delay': 40+i,
'tx_delay': 50+i,
'tx_dbm': 60+i
},
'mac_addr': '00:0A:45:00:00:%02x' % i,
}
# radio units configuration in enb.cfg
def test_rf_cfg_ru(t):
assertMatch(t, t.rf_cfg['rf_driver'], dict(
name='sdr',
args='dev0=/dev/sdr0@1,dev1=/dev/sdr0@2,dev2=/dev/sdr0@3,dev3=/dev/sdr0@4',
cpri_mapping='hw,hw,hw,hw',
cpri_mult='4,4,4,4',
cpri_rx_delay='41,42,43,44',
cpri_tx_delay='51,52,53,54',
cpri_tx_dbm='61,62,63,64',
))
# RU configuration in cu_config.xml
def test_ru_cu_config_xml(t):
def uctx(rf_mode, cell_type, dl_arfcn, ul_arfcn, bw, dl_freq, ul_freq, tx_gain, rx_gain):
return {
'tx-array-carriers': {
'rw-duplex-scheme': rf_mode,
'rw-type': cell_type,
'absolute-frequency-center': '%d' % dl_arfcn,
'center-of-channel-bandwidth': '%d' % dl_freq,
'channel-bandwidth': '%d' % bw,
'gain': '%d' % tx_gain,
'active': 'ACTIVE',
},
'rx-array-carriers': {
'absolute-frequency-center': '%d' % ul_arfcn,
'center-of-channel-bandwidth': '%d' % ul_freq,
'channel-bandwidth': '%d' % bw,
# XXX no rx_gain
'active': 'ACTIVE',
},
}
_ = t._test_ru_cu_config_xml
# rf_mode ctype dl_arfcn ul_arfcn bw dl_freq ul_freq txg rxg
_(1, uctx('FDD', 'LTE', 100, 18100, 5000000, 2120000000, 1930000000, 11, 21))
_(2, uctx('TDD', 'LTE', 40200, 40200, 10000000, 2551000000, 2551000000, 12, 22))
_(3, uctx('FDD', 'NR', 300300, 290700, 15000000, 1501500000, 1453500000, 13, 23))
_(4, uctx('TDD', 'NR', 470400, 470400, 20000000, 2352000000, 2352000000, 14, 24))
def _test_ru_cu_config_xml(t, i, uctx):
cu_xml = t.ipath('etc/%s' % xbuildout.encode('%s-cu_config.xml' % t.ref('RU%d' % i)))
with open(cu_xml, 'r') as f:
cu = f.read()
cu = xmltodict.parse(cu)
assertMatch(t, cu, {
'xc:config': {
'user-plane-configuration': {
'tx-endpoints': [
{'name': 'TXA0P00C00', 'e-axcid': {'eaxc-id': '0'}},
{'name': 'TXA0P00C01', 'e-axcid': {'eaxc-id': '1'}},
{'name': 'TXA0P01C00', 'e-axcid': {'eaxc-id': '2'}},
{'name': 'TXA0P01C01', 'e-axcid': {'eaxc-id': '3'}},
],
'tx-links': [
{'name': 'TXA0P00C00', 'tx-endpoint': 'TXA0P00C00'},
{'name': 'TXA0P00C01', 'tx-endpoint': 'TXA0P00C01'},
{'name': 'TXA0P01C00', 'tx-endpoint': 'TXA0P01C00'},
{'name': 'TXA0P01C01', 'tx-endpoint': 'TXA0P01C01'},
],
'rx-endpoints': [
{'name': 'RXA0P00C00', 'e-axcid': {'eaxc-id': '0'}},
{'name': 'PRACH0P00C00', 'e-axcid': {'eaxc-id': '8'}},
{'name': 'RXA0P00C01', 'e-axcid': {'eaxc-id': '1'}},
{'name': 'PRACH0P00C01', 'e-axcid': {'eaxc-id': '24'}},
],
'rx-links': [
{'name': 'RXA0P00C00', 'rx-endpoint': 'RXA0P00C00'},
{'name': 'PRACH0P00C00', 'rx-endpoint': 'PRACH0P00C00'},
{'name': 'RXA0P00C01', 'rx-endpoint': 'RXA0P00C01'},
{'name': 'PRACH0P00C01', 'rx-endpoint': 'PRACH0P00C01'},
],
} | uctx
}
})
# RU configuration in cu_inactive_config.xml
def test_ru_cu_inactive_config_xml(t):
def uctx(rf_mode, cell_type, dl_arfcn, ul_arfcn, bw, dl_freq, ul_freq, tx_gain, rx_gain):
return {
'tx-array-carriers': {
'rw-duplex-scheme': rf_mode,
'rw-type': cell_type,
'absolute-frequency-center': '%d' % dl_arfcn,
'center-of-channel-bandwidth': '%d' % dl_freq,
'channel-bandwidth': '%d' % bw,
'gain': '%d' % tx_gain,
'active': 'INACTIVE',
},
'rx-array-carriers': {
'absolute-frequency-center': '%d' % ul_arfcn,
'center-of-channel-bandwidth': '%d' % ul_freq,
'channel-bandwidth': '%d' % bw,
# XXX no rx_gain
'active': 'INACTIVE',
},
}
_ = t._test_ru_cu_inactive_config_xml
# rf_mode ctype dl_arfcn ul_arfcn bw dl_freq ul_freq txg rxg
_(1, uctx('FDD', 'LTE', 100, 18100, 5000000, 2120000000, 1930000000, 11, 21))
_(2, uctx('TDD', 'LTE', 40200, 40200, 10000000, 2551000000, 2551000000, 12, 22))
_(3, uctx('FDD', 'NR', 300300, 290700, 15000000, 1501500000, 1453500000, 13, 23))
_(4, uctx('TDD', 'NR', 470400, 470400, 20000000, 2352000000, 2352000000, 14, 24))
def _test_ru_cu_inactive_config_xml(t, i, uctx):
cu_xml = t.ipath('etc/%s' % xbuildout.encode('%s-cu_inactive_config.xml' % t.ref('RU%d' % i)))
with open(cu_xml, 'r') as f:
cu = f.read()
cu = xmltodict.parse(cu)
assertMatch(t, cu, {
'xc:config': {
'user-plane-configuration': {
'tx-endpoints': [
{'name': 'TXA0P00C00', 'e-axcid': {'eaxc-id': '0'}},
{'name': 'TXA0P00C01', 'e-axcid': {'eaxc-id': '1'}},
{'name': 'TXA0P01C00', 'e-axcid': {'eaxc-id': '2'}},
{'name': 'TXA0P01C01', 'e-axcid': {'eaxc-id': '3'}},
],
'tx-links': [
{'name': 'TXA0P00C00', 'tx-endpoint': 'TXA0P00C00'},
{'name': 'TXA0P00C01', 'tx-endpoint': 'TXA0P00C01'},
{'name': 'TXA0P01C00', 'tx-endpoint': 'TXA0P01C00'},
{'name': 'TXA0P01C01', 'tx-endpoint': 'TXA0P01C01'},
],
'rx-endpoints': [
{'name': 'RXA0P00C00', 'e-axcid': {'eaxc-id': '0'}},
{'name': 'PRACH0P00C00', 'e-axcid': {'eaxc-id': '8'}},
{'name': 'RXA0P00C01', 'e-axcid': {'eaxc-id': '1'}},
{'name': 'PRACH0P00C01', 'e-axcid': {'eaxc-id': '24'}},
],
'rx-links': [
{'name': 'RXA0P00C00', 'rx-endpoint': 'RXA0P00C00'},
{'name': 'PRACH0P00C00', 'rx-endpoint': 'PRACH0P00C00'},
{'name': 'RXA0P00C01', 'rx-endpoint': 'RXA0P00C01'},
{'name': 'PRACH0P00C01', 'rx-endpoint': 'PRACH0P00C01'},
],
} | uctx
}
})
# Sunwave4 is mixin to verify Sunwave driver wrt all LTE/NR x FDD/TDD modes. # Sunwave4 is mixin to verify Sunwave driver wrt all LTE/NR x FDD/TDD modes.
class Sunwave4: class Sunwave4:
@classmethod @classmethod
...@@ -699,35 +534,35 @@ class Sunwave4: ...@@ -699,35 +534,35 @@ class Sunwave4:
)) ))
# RUMultiType4 is mixin to verify that different RU types can be used at the same time. # RUMultiType4 is mixin to verify that different RU types can be used at the same time.
class RUMultiType4: # Due to only one type of RU being supported in the SR, the test is currently not applicable.
# ENB does not support mixing SDR + CPRI - verify only with CPRI-based units #class RUMultiType4:
# see https://support.amarisoft.com/issues/26021 for details # # ENB does not support mixing SDR + CPRI - verify only with CPRI-based units
@classmethod # # see https://support.amarisoft.com/issues/26021 for details
def RUcfg(cls, i): # @classmethod
assert 1 <= i <= 4, i # def RUcfg(cls, i):
if i in (1,2): # assert 1 <= i <= 4, i
return Lopcomm4.RUcfg(i) # if i in (1,2):
else: # return SDR4.RUcfg(i)
return Sunwave4.RUcfg(i) # else:
# return Sunwave4.RUcfg(i)
# radio units configuration in enb.cfg #
def test_rf_cfg_ru(t): # # radio units configuration in enb.cfg
assertMatch(t, t.rf_cfg['rf_driver'], dict( # def test_rf_cfg_ru(t):
name='sdr', # assertMatch(t, t.rf_cfg['rf_driver'], dict(
args='dev0=/dev/sdr0@1,dev1=/dev/sdr0@2,dev2=/dev/sdr1@3,dev3=/dev/sdr1@4', # name='sdr',
cpri_mapping='hw,hw,bf1,bf1', # args='dev0=/dev/sdr0@1,dev1=/dev/sdr0@2,dev2=/dev/sdr1@3,dev3=/dev/sdr1@4',
cpri_mult='4,4,5,5', # cpri_mapping='hw,hw,bf1,bf1',
cpri_rx_delay='41,42,143,144', # cpri_mult='4,4,5,5',
cpri_tx_delay='51,52,153,154', # cpri_rx_delay='41,42,143,144',
cpri_tx_dbm='61,62,163,164', # cpri_tx_delay='51,52,153,154',
)) # cpri_tx_dbm='61,62,163,164',
# ))
# instantiate eNB tests # instantiate eNB tests
class TestENB_SDR4 (ENBTestCase4, SDR4): pass class TestENB_SDR4 (ENBTestCase4, SDR4): pass
class TestENB_Lopcomm4 (ENBTestCase4, Lopcomm4): pass
class TestENB_Sunwave4 (ENBTestCase4, Sunwave4): pass class TestENB_Sunwave4 (ENBTestCase4, Sunwave4): pass
class TestENB_RUMultiType4(ENBTestCase4, RUMultiType4): pass # class TestENB_RUMultiType4(ENBTestCase4, RUMultiType4): pass
# ---- UEsim ---- # ---- UEsim ----
...@@ -844,9 +679,8 @@ class UEsimTestCase4(RFTestCase4): ...@@ -844,9 +679,8 @@ class UEsimTestCase4(RFTestCase4):
# instantiate UEsim tests # instantiate UEsim tests
class TestUEsim_SDR4 (UEsimTestCase4, SDR4): pass class TestUEsim_SDR4 (UEsimTestCase4, SDR4): pass
class TestUEsim_Lopcomm4 (UEsimTestCase4, Lopcomm4): pass
class TestUEsim_Sunwave4 (UEsimTestCase4, Sunwave4): pass class TestUEsim_Sunwave4 (UEsimTestCase4, Sunwave4): pass
class TestUEsim_RUMultiType4(UEsimTestCase4, RUMultiType4): pass # class TestUEsim_RUMultiType4(UEsimTestCase4, RUMultiType4): pass
# ---- misc ---- # ---- misc ----
......
...@@ -410,11 +410,6 @@ eggs += ...@@ -410,11 +410,6 @@ eggs +=
# custom eggs pre-installed, not our special python interpreter. # custom eggs pre-installed, not our special python interpreter.
interpreter = python_for_test interpreter = python_for_test
# patches for eggs
patch-binary = ${patch:location}/bin/patch
PyPDF2-patches = ${:_profile_base_location_}/../../component/egg-patch/PyPDF2/0001-Custom-implementation-of-warnings.formatwarning-remo.patch#d25bb0f5dde7f3337a0a50c2f986f5c8
PyPDF2-patch-options = -p1
[eggs/scripts] [eggs/scripts]
recipe = zc.recipe.egg recipe = zc.recipe.egg
eggs = ${python-interpreter:eggs} eggs = ${python-interpreter:eggs}
...@@ -514,6 +509,7 @@ Pillow = 10.2.0+SlapOSPatched001 ...@@ -514,6 +509,7 @@ Pillow = 10.2.0+SlapOSPatched001
forcediphttpsadapter = 1.0.1 forcediphttpsadapter = 1.0.1
image = 1.5.25 image = 1.5.25
plantuml = 0.3.0:whl plantuml = 0.3.0:whl
pypdf = 3.6.0:whl
pysftp = 0.2.9 pysftp = 0.2.9
requests-toolbelt = 0.8.0 requests-toolbelt = 0.8.0
testfixtures = 6.11.0 testfixtures = 6.11.0
...@@ -522,9 +518,6 @@ paho-mqtt = 1.5.0 ...@@ -522,9 +518,6 @@ paho-mqtt = 1.5.0
pcpp = 1.30 pcpp = 1.30
xmltodict = 0.13.0 xmltodict = 0.13.0
# Patched eggs
PyPDF2 = 1.26.0+SlapOSPatched001
# Test Suite: SlapOS.SoftwareReleases.IntegrationTest-Master.Python2 ran at 2022/09/08 02:05:35.783873 UTC # Test Suite: SlapOS.SoftwareReleases.IntegrationTest-Master.Python2 ran at 2022/09/08 02:05:35.783873 UTC
# 2 failures, 0 errors, 1037 total, status: FAIL # 2 failures, 0 errors, 1037 total, status: FAIL
......
...@@ -353,7 +353,7 @@ sgmllib3k = 1.0.0 ...@@ -353,7 +353,7 @@ sgmllib3k = 1.0.0
simplegeneric = 0.8.1 simplegeneric = 0.8.1
singledispatch = 3.4.0.3 singledispatch = 3.4.0.3
six = 1.16.0 six = 1.16.0
slapos.cookbook = 1.0.360 slapos.cookbook = 1.0.365
slapos.core = 1.12.0 slapos.core = 1.12.0
slapos.extension.shared = 1.0 slapos.extension.shared = 1.0
slapos.libnetworkcache = 0.25 slapos.libnetworkcache = 0.25
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment